




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


lowercase__     						=						pytest.mark.integration






@require_faiss
class lowerCAmelCase__	(     lowercase						):






			'''simple docstring'''



			def 					A_    (  self  ):
							_lowerCamelCase     :			int    =    Dataset.from_dict({'filename': ['my_name-train' + '_' + str(lowercase  ) for x in np.arange(30  ).tolist()]}  )
							return dset
			def 					A_    (  self  ):
							import faiss

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

							_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[int]    =    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

							_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=lowercase  ) 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     :			Dict    =    dset.get_nearest_examples('vecs2'  ,      np.ones(5  ,      dtype=np.floataa  )  )
							self.assertEqual(examples['filename'][0]  ,      'my_name-train_29'  )
			def 					A_    (  self  ):
							_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(lowercase  ,      partial(dset.get_nearest_examples  ,      'vecs2'  ,      np.ones(5  ,      dtype=np.floataa  )  )  )






			def 					A_    (  self  ):
							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     :			Tuple    =    {'acknowledged': True}
											mocked_bulk.return_value([(True, None)] * 30  )
											_lowerCamelCase     :			List[Any]    =    {'hits': {'hits': [{'_score': 1, '_id': 29}]}}
											_lowerCamelCase     :			Optional[int]    =    Elasticsearch()

											dset.add_elasticsearch_index('filename'  ,      es_client=lowercase  )
											_lowerCamelCase, _lowerCamelCase     :			Union[str, Any]    =    dset.get_nearest_examples('filename'  ,      'my_name-train_29'  )
											self.assertEqual(examples['filename'][0]  ,      'my_name-train_29'  )






@require_faiss
class lowerCAmelCase__	(     lowercase						):






			'''simple docstring'''



			def 					A_    (  self  ):
							import faiss

							_lowerCamelCase     :			str    =    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     :			Dict    =    np.zeros(5  ,      dtype=np.floataa  )
							_lowerCamelCase     :			Dict    =    1
							_lowerCamelCase, _lowerCamelCase     :			int    =    index.search(lowercase  )
							self.assertRaises(lowercase  ,      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(lowercase  )
							self.assertRaises(lowercase  ,      index.search_batch  ,      queries[0]  )
							_lowerCamelCase     :			List[str]    =    [scores[0] for scores in total_scores]
							_lowerCamelCase     :			List[Any]    =    [indices[0] for indices in total_indices]
							self.assertGreater(np.min(lowercase  )  ,      0  )
							self.assertListEqual([4, 3, 2, 1, 0]  ,      lowercase  )
			def 					A_    (  self  ):
							import faiss

							_lowerCamelCase     :			Tuple    =    FaissIndex(string_factory='Flat'  )
							index.add_vectors(np.eye(5  ,      dtype=np.floataa  )  )
							self.assertIsInstance(index.faiss_index  ,      faiss.IndexFlat  )
							_lowerCamelCase     :			int    =    FaissIndex(string_factory='LSH'  )
							index.add_vectors(np.eye(5  ,      dtype=np.floataa  )  )
							self.assertIsInstance(index.faiss_index  ,      faiss.IndexLSH  )
							with self.assertRaises(lowercase  ):
											_lowerCamelCase     :			List[Any]    =    FaissIndex(string_factory='Flat'  ,      custom_index=faiss.IndexFlat(5  )  )
			def 					A_    (  self  ):
							import faiss

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






			def 					A_    (  self  ):
							import faiss

							_lowerCamelCase     :			Tuple    =    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=lowercase  ) as tmp_file:
											index.save(tmp_file.name  )
											_lowerCamelCase     :			Union[str, Any]    =    FaissIndex.load(tmp_file.name  )
							os.unlink(tmp_file.name  )

							_lowerCamelCase     :			Tuple    =    np.zeros(5  ,      dtype=np.floataa  )
							_lowerCamelCase     :			Optional[int]    =    1
							_lowerCamelCase, _lowerCamelCase     :			Tuple    =    index.search(lowercase  )
							self.assertGreater(scores[0]  ,      0  )
							self.assertEqual(indices[0]  ,      1  )







@require_faiss
def      _snake_case			(				lowercase__		):
				import faiss

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

				_lowerCamelCase     :			Dict    =    'index.faiss'
				_lowerCamelCase     :			Optional[int]    =    f'''mock://{index_name}'''
				index.save(lowercase__							,    storage_options=mockfs.storage_options		)
				_lowerCamelCase     :			Dict    =    FaissIndex.load(lowercase__							,    storage_options=mockfs.storage_options		)

				_lowerCamelCase     :			Union[str, Any]    =    np.zeros(5							,    dtype=np.floataa		)
				_lowerCamelCase     :			Any    =    1
				_lowerCamelCase, _lowerCamelCase     :			List[Any]    =    index.search(lowercase__		)
				assert scores[0] > 0
				assert indices[0] == 1






@require_elasticsearch
class lowerCAmelCase__	(     lowercase						):






			'''simple docstring'''



			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:
											_lowerCamelCase     :			Tuple    =    Elasticsearch()
											_lowerCamelCase     :			List[Any]    =    {'acknowledged': True}
											_lowerCamelCase     :			Optional[Any]    =    ElasticSearchIndex(es_client=lowercase  )
											mocked_bulk.return_value([(True, None)] * 3  )
											index.add_documents(['foo', 'bar', 'foobar']  )

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

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

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

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





import json
import logging
import os
import sys
from pathlib import Path

import finetune_rag

from transformers.file_utils import is_apex_available
from transformers.testing_utils import (
    TestCasePlus,
    execute_subprocess_async,
    require_ray,
    require_torch_gpu,
    require_torch_multi_gpu,
)


logging.basicConfig(level=logging.DEBUG)
SCREAMING_SNAKE_CASE_:     Optional[int]  =logging.getLogger()

SCREAMING_SNAKE_CASE_:     Union[str, Any]  =logging.StreamHandler(sys.stdout)
logger.addHandler(stream_handler)




class        __A      ( UpperCamelCase__	):



    def 					_lowercase						(self    :    Optional[Any]   ,			__a    :    str ):
         os.makedirs(__a   ,			exist_ok=__a )
         UpperCAmelCase_           =   {"source": "What is love ?", "target": "life"}
         UpperCAmelCase_           =   {"train": 12, "val": 2, "test": 2}
         for split in ["train", "test", "val"]:
              for field in ["source", "target"]:
                   UpperCAmelCase_           =   "\n".join([contents[field]] * n_lines[split] )
                   with open(os.path.join(__a   ,			f"""{split}.{field}""" )   ,			"w" ) as f:
                        f.write(__a )



    def 					_lowercase						(self    :    Optional[int]   ,			__a    :    int   ,			__a    :    str = "pytorch" ):
         UpperCAmelCase_           =   self.get_auto_remove_tmp_dir()
         UpperCAmelCase_           =   os.path.join(__a   ,			"output" )
         UpperCAmelCase_           =   os.path.join(__a   ,			"data" )
         self._create_dummy_data(data_dir=__a )

         UpperCAmelCase_           =   f"""
                --data_dir {data_dir} \
                --output_dir {output_dir} \
                --model_name_or_path facebook/rag-sequence-base \
                --model_type rag_sequence \
                --do_train \
                --do_predict \
                --n_val -1 \
                --val_check_interval 1.0 \
                --train_batch_size 2 \
                --eval_batch_size 1 \
                --max_source_length 25 \
                --max_target_length 25 \
                --val_max_target_length 25 \
                --test_max_target_length 25 \
                --label_smoothing 0.1 \
                --dropout 0.1 \
                --attention_dropout 0.1 \
                --weight_decay 0.001 \
                --adam_epsilon 1e-08 \
                --max_grad_norm 0.1 \
                --lr_scheduler polynomial \
                --learning_rate 3e-04 \
                --num_train_epochs 1 \
                --warmup_steps 4 \
                --gradient_accumulation_steps 1 \
                --distributed-port 8787 \
                --use_dummy_dataset 1 \
                --distributed_retriever {distributed_retriever} \
            """.split()

         if gpus > 0:
              testargs.append(f"""--gpus={gpus}""" )
              if is_apex_available():
                   testargs.append("--fp16" )
         else:
              testargs.append("--gpus=0" )
              testargs.append("--distributed_backend=ddp_cpu" )
              testargs.append("--num_processes=2" )

         UpperCAmelCase_           =   [sys.executable, str(Path(finetune_rag.__file__ ).resolve() )] + testargs
         execute_subprocess_async(__a   ,			env=self.get_env() )

         UpperCAmelCase_           =   os.path.join(__a   ,			"metrics.json" )
         with open(__a ) as f:
              UpperCAmelCase_           =   json.load(__a )
         return result



    @require_torch_gpu
    def 					_lowercase						(self    :    Optional[int] ):
         UpperCAmelCase_           =   self._run_finetune(gpus=1 )
         self.assertGreaterEqual(result["test"][0]["test_avg_em"]   ,			0.2 )



    @require_torch_multi_gpu
    def 					_lowercase						(self    :    Dict ):
         UpperCAmelCase_           =   self._run_finetune(gpus=2 )
         self.assertGreaterEqual(result["test"][0]["test_avg_em"]   ,			0.2 )



    @require_torch_gpu
    @require_ray
    def 					_lowercase						(self    :    Optional[int] ):
         UpperCAmelCase_           =   self._run_finetune(gpus=1   ,			distributed_retriever="ray" )
         self.assertGreaterEqual(result["test"][0]["test_avg_em"]   ,			0.2 )






    @require_torch_multi_gpu
    @require_ray
    def 					_lowercase						(self    :    Any ):
         UpperCAmelCase_           =   self._run_finetune(gpus=1   ,			distributed_retriever="ray" )
         self.assertGreaterEqual(result["test"][0]["test_avg_em"]   ,			0.2 )




