

"""simple docstring"""





import json
import os
import shutil
import sys
import tempfile
import unittest
from unittest import TestCase
from unittest.mock import patch

import faiss
import numpy as np
from datasets import Dataset

from transformers import BartConfig, BartTokenizer, DPRConfig, DPRQuestionEncoderTokenizer, RagConfig
from transformers.file_utils import is_datasets_available, is_faiss_available, is_psutil_available, is_torch_available
from transformers.integrations import is_ray_available
from transformers.models.bert.tokenization_bert import VOCAB_FILES_NAMES as DPR_VOCAB_FILES_NAMES
from transformers.models.rag.retrieval_rag import CustomHFIndex, RagRetriever
from transformers.models.roberta.tokenization_roberta import VOCAB_FILES_NAMES as BART_VOCAB_FILES_NAMES
from transformers.testing_utils import require_ray


sys.path.append(os.path.join(os.getcwd()))  # noqa: E402 # noqa: E402 # isort:skip

if is_torch_available():
							from distributed_pytorch_retriever import RagPyTorchDistributedRetriever  # noqa: E402 # isort:skip
else:
							__UpperCamelCase						:   Union[str, Any]							   =	None

if is_ray_available():
							import ray  # noqa: E402 # isort:skip
							from distributed_ray_retriever import RagRayDistributedRetriever, RayRetriever  # noqa: E402 # isort:skip
else:
							__UpperCamelCase						:   int							   =	None
							__UpperCamelCase						:   Optional[Any]							   =	None
							__UpperCamelCase						:   str							   =	None





def 						__SCREAMING_SNAKE_CASE						(   A_				):



	if not (is_datasets_available() and is_faiss_available() and is_psutil_available()):
		lowerCAmelCase__			:		str						       =			unittest.skip('''test requires Datasets, Faiss, psutil'''				)(A_				)
	return test_case







@require_distributed_retrieval
class        SCREAMING_SNAKE_CASE      (	lowerCAmelCase__ ):






					"""simple docstring"""



					def  __lowerCAmelCase				( self    :       Union[str, Any]	):
						lowerCAmelCase__			:		str						       =			tempfile.mkdtemp()
						lowerCAmelCase__			:		Tuple						       =			8

						# DPR tok
						lowerCAmelCase__			:		Tuple						       =			[
						    '''[UNK]''',
						    '''[CLS]''',
						    '''[SEP]''',
						    '''[PAD]''',
						    '''[MASK]''',
						    '''want''',
						    '''##want''',
						    '''##ed''',
						    '''wa''',
						    '''un''',
						    '''runn''',
						    '''##ing''',
						    ''',''',
						    '''low''',
						    '''lowest''',
						]
						lowerCAmelCase__			:		Dict						       =			os.path.join(self.tmpdirname     ,'''dpr_tokenizer'''	)
						os.makedirs(lowerCAmelCase_     ,exist_ok=lowerCAmelCase_	)
						lowerCAmelCase__			:		Optional[Any]						       =			os.path.join(lowerCAmelCase_     ,DPR_VOCAB_FILES_NAMES['''vocab_file''']	)
						with open(self.vocab_file     ,'''w'''     ,encoding='''utf-8'''	) as vocab_writer:
							vocab_writer.write(''''''.join([x + '''\n''' for x in vocab_tokens]	)	)

						# BART tok
						lowerCAmelCase__			:		str						       =			[
						    '''l''',
						    '''o''',
						    '''w''',
						    '''e''',
						    '''r''',
						    '''s''',
						    '''t''',
						    '''i''',
						    '''d''',
						    '''n''',
						    '''\u0120''',
						    '''\u0120l''',
						    '''\u0120n''',
						    '''\u0120lo''',
						    '''\u0120low''',
						    '''er''',
						    '''\u0120lowest''',
						    '''\u0120newer''',
						    '''\u0120wider''',
						    '''<unk>''',
						]
						lowerCAmelCase__			:		List[str]						       =			dict(zip(lowerCAmelCase_     ,range(len(lowerCAmelCase_	)	)	)	)
						lowerCAmelCase__			:		List[str]						       =			['''#version: 0.2''', '''\u0120 l''', '''\u0120l o''', '''\u0120lo w''', '''e r''', '''''']
						lowerCAmelCase__			:		List[Any]						       =			{'''unk_token''': '''<unk>'''}

						lowerCAmelCase__			:		List[Any]						       =			os.path.join(self.tmpdirname     ,'''bart_tokenizer'''	)
						os.makedirs(lowerCAmelCase_     ,exist_ok=lowerCAmelCase_	)
						lowerCAmelCase__			:		Optional[Any]						       =			os.path.join(lowerCAmelCase_     ,BART_VOCAB_FILES_NAMES['''vocab_file''']	)
						lowerCAmelCase__			:		Any						       =			os.path.join(lowerCAmelCase_     ,BART_VOCAB_FILES_NAMES['''merges_file''']	)
						with open(self.vocab_file     ,'''w'''     ,encoding='''utf-8'''	) as fp:
							fp.write(json.dumps(lowerCAmelCase_	) + '''\n'''	)
						with open(self.merges_file     ,'''w'''     ,encoding='''utf-8'''	) as fp:
							fp.write('''\n'''.join(lowerCAmelCase_	)	)







					def  __lowerCAmelCase				( self    :       str	):
						return DPRQuestionEncoderTokenizer.from_pretrained(os.path.join(self.tmpdirname     ,'''dpr_tokenizer'''	)	)







					def  __lowerCAmelCase				( self    :       Dict	):
						return BartTokenizer.from_pretrained(os.path.join(self.tmpdirname     ,'''bart_tokenizer'''	)	)







					def  __lowerCAmelCase				( self    :       List[Any]	):
						shutil.rmtree(self.tmpdirname	)







					def  __lowerCAmelCase				( self    :       Any	):
						lowerCAmelCase__			:		str						       =			Dataset.from_dict(
						    {
						        '''id''': ['''0''', '''1'''],
						        '''text''': ['''foo''', '''bar'''],
						        '''title''': ['''Foo''', '''Bar'''],
						        '''embeddings''': [np.ones(self.retrieval_vector_size	), 2 * np.ones(self.retrieval_vector_size	)],
						    }	)
						dataset.add_faiss_index('''embeddings'''     ,string_factory='''Flat'''     ,metric_type=faiss.METRIC_INNER_PRODUCT	)
						return dataset







					def  __lowerCAmelCase				( self    :       List[str]     ,lowercase_    :       bool     ,lowercase_    :       List[str]=1_2_3_4_5	):
						lowerCAmelCase__			:		Dict						       =			self.get_dummy_dataset()
						lowerCAmelCase__			:		Dict						       =			RagConfig(
						    retrieval_vector_size=self.retrieval_vector_size     ,question_encoder=DPRConfig().to_dict()     ,generator=BartConfig().to_dict()     ,)
						with patch('''transformers.models.rag.retrieval_rag.load_dataset'''	) as mock_load_dataset:
							lowerCAmelCase__			:		Optional[int]						       =			dataset
							lowerCAmelCase__			:		Any						       =			RagPyTorchDistributedRetriever(
							    lowerCAmelCase_     ,question_encoder_tokenizer=self.get_dpr_tokenizer()     ,generator_tokenizer=self.get_bart_tokenizer()     ,)
							if init_retrieval:
								retriever.init_retrieval(lowerCAmelCase_	)
						return retriever







					def  __lowerCAmelCase				( self    :       Optional[int]     ,lowercase_    :       bool	):
						ray.init(local_mode=lowerCAmelCase_	)
						lowerCAmelCase__			:		str						       =			RagConfig(
						    retrieval_vector_size=self.retrieval_vector_size     ,question_encoder=DPRConfig().to_dict()     ,generator=BartConfig().to_dict()     ,)
						lowerCAmelCase__			:		Dict						       =			ray.remote(lowerCAmelCase_	)
						lowerCAmelCase__			:		List[Any]						       =			[remote_cls.remote() for _ in range(1	)]
						with patch('''transformers.models.rag.retrieval_rag.load_dataset'''	) as mock_load_dataset:
							lowerCAmelCase__			:		Optional[int]						       =			self.get_dummy_dataset()
							lowerCAmelCase__			:		List[Any]						       =			RagRayDistributedRetriever(
							    lowerCAmelCase_     ,question_encoder_tokenizer=self.get_dpr_tokenizer()     ,generator_tokenizer=self.get_bart_tokenizer()     ,retrieval_workers=lowerCAmelCase_     ,)
							if init_retrieval:
								retriever.init_retrieval()
						return retriever







					def  __lowerCAmelCase				( self    :       Tuple     ,lowercase_    :       bool     ,lowercase_    :       bool     ,lowercase_    :       List[str]=1_2_3_4_5	):
						lowerCAmelCase__			:		List[str]						       =			self.get_dummy_dataset()
						lowerCAmelCase__			:		Optional[Any]						       =			RagConfig(
						    retrieval_vector_size=self.retrieval_vector_size     ,question_encoder=DPRConfig().to_dict()     ,generator=BartConfig().to_dict()     ,index_name='''custom'''     ,)
						if from_disk:
							lowerCAmelCase__			:		List[Any]						       =			os.path.join(self.tmpdirname     ,'''dataset'''	)
							lowerCAmelCase__			:		Dict						       =			os.path.join(self.tmpdirname     ,'''index.faiss'''	)
							dataset.get_index('''embeddings'''	).save(os.path.join(self.tmpdirname     ,'''index.faiss'''	)	)
							dataset.drop_index('''embeddings'''	)
							dataset.save_to_disk(os.path.join(self.tmpdirname     ,'''dataset'''	)	)
							del dataset
							lowerCAmelCase__			:		str						       =			RagPyTorchDistributedRetriever(
							    lowerCAmelCase_     ,question_encoder_tokenizer=self.get_dpr_tokenizer()     ,generator_tokenizer=self.get_bart_tokenizer()     ,)
						else:
							lowerCAmelCase__			:		Union[str, Any]						       =			RagPyTorchDistributedRetriever(
							    lowerCAmelCase_     ,question_encoder_tokenizer=self.get_dpr_tokenizer()     ,generator_tokenizer=self.get_bart_tokenizer()     ,index=CustomHFIndex(config.retrieval_vector_size     ,lowerCAmelCase_	)     ,)
						if init_retrieval:
							retriever.init_retrieval(lowerCAmelCase_	)
						return retriever







					def  __lowerCAmelCase				( self    :       List[Any]     ,lowercase_    :       bool     ,lowercase_    :       bool	):
						ray.init(local_mode=lowerCAmelCase_	)
						lowerCAmelCase__			:		Any						       =			self.get_dummy_dataset()
						lowerCAmelCase__			:		str						       =			RagConfig(
						    retrieval_vector_size=self.retrieval_vector_size     ,question_encoder=DPRConfig().to_dict()     ,generator=BartConfig().to_dict()     ,index_name='''custom'''     ,)
						lowerCAmelCase__			:		Optional[int]						       =			ray.remote(lowerCAmelCase_	)
						lowerCAmelCase__			:		int						       =			[remote_cls.remote() for _ in range(1	)]
						if from_disk:
							lowerCAmelCase__			:		str						       =			os.path.join(self.tmpdirname     ,'''dataset'''	)
							lowerCAmelCase__			:		List[str]						       =			os.path.join(self.tmpdirname     ,'''index.faiss'''	)
							dataset.get_index('''embeddings'''	).save(os.path.join(self.tmpdirname     ,'''index.faiss'''	)	)
							dataset.drop_index('''embeddings'''	)
							dataset.save_to_disk(os.path.join(self.tmpdirname     ,'''dataset'''	)	)
							del dataset
							lowerCAmelCase__			:		Dict						       =			RagRayDistributedRetriever(
							    lowerCAmelCase_     ,question_encoder_tokenizer=self.get_dpr_tokenizer()     ,generator_tokenizer=self.get_bart_tokenizer()     ,retrieval_workers=lowerCAmelCase_     ,index=CustomHFIndex.load_from_disk(
							        vector_size=config.retrieval_vector_size     ,dataset_path=config.passages_path     ,index_path=config.index_path     ,)     ,)
						else:
							lowerCAmelCase__			:		Dict						       =			RagRayDistributedRetriever(
							    lowerCAmelCase_     ,question_encoder_tokenizer=self.get_dpr_tokenizer()     ,generator_tokenizer=self.get_bart_tokenizer()     ,retrieval_workers=lowerCAmelCase_     ,index=CustomHFIndex(config.retrieval_vector_size     ,lowerCAmelCase_	)     ,)
						if init_retrieval:
							retriever.init_retrieval()
						return retriever







					def  __lowerCAmelCase				( self    :       Optional[Any]     ,lowercase_    :       RagRetriever     ,lowercase_    :       np.array     ,lowercase_    :       int	):
						lowerCAmelCase__       ,lowerCAmelCase__       ,lowerCAmelCase__			:		Union[str, Any]						       =			retriever.retrieve(lowerCAmelCase_     ,n_docs=lowerCAmelCase_	)
						self.assertEqual(retrieved_doc_embeds.shape     ,(2, n_docs, self.retrieval_vector_size)	)
						self.assertEqual(len(lowerCAmelCase_	)     ,2	)
						self.assertEqual(sorted(doc_dicts[0]	)     ,['''embeddings''', '''id''', '''text''', '''title''']	)
						self.assertEqual(len(doc_dicts[0]['''id''']	)     ,lowerCAmelCase_	)
						self.assertEqual(doc_dicts[0]['''id'''][0]     ,'''1'''	)  # max inner product is reached with second doc
						self.assertEqual(doc_dicts[1]['''id'''][0]     ,'''0'''	)  # max inner product is reached with first doc
						self.assertListEqual(doc_ids.tolist()     ,[[1], [0]]	)







					def  __lowerCAmelCase				( self    :       Optional[int]	):
						lowerCAmelCase__			:		List[str]						       =			1
						lowerCAmelCase__			:		List[str]						       =			np.array(
						    [np.ones(self.retrieval_vector_size	), -np.ones(self.retrieval_vector_size	)]     ,dtype=np.floataa	)

						self.distributed_retriever_check(
						    self.get_dummy_pytorch_distributed_retriever(init_retrieval=lowerCAmelCase_	)     ,lowerCAmelCase_     ,lowerCAmelCase_	)







					def  __lowerCAmelCase				( self    :       List[str]	):
						lowerCAmelCase__			:		Optional[Any]						       =			1
						lowerCAmelCase__			:		Tuple						       =			np.array(
						    [np.ones(self.retrieval_vector_size	), -np.ones(self.retrieval_vector_size	)]     ,dtype=np.floataa	)

						self.distributed_retriever_check(
						    self.get_dummy_custom_hf_index_pytorch_retriever(init_retrieval=lowerCAmelCase_     ,from_disk=lowerCAmelCase_	)     ,lowerCAmelCase_     ,lowerCAmelCase_     ,)







					def  __lowerCAmelCase				( self    :       Union[str, Any]	):
						lowerCAmelCase__			:		str						       =			1
						lowerCAmelCase__			:		Union[str, Any]						       =			np.array(
						    [np.ones(self.retrieval_vector_size	), -np.ones(self.retrieval_vector_size	)]     ,dtype=np.floataa	)

						self.distributed_retriever_check(
						    self.get_dummy_custom_hf_index_pytorch_retriever(init_retrieval=lowerCAmelCase_     ,from_disk=lowerCAmelCase_	)     ,lowerCAmelCase_     ,lowerCAmelCase_     ,)







					@require_ray
					def  __lowerCAmelCase				( self    :       Dict	):
						lowerCAmelCase__			:		Union[str, Any]						       =			1
						lowerCAmelCase__			:		str						       =			np.array(
						    [np.ones(self.retrieval_vector_size	), -np.ones(self.retrieval_vector_size	)]     ,dtype=np.floataa	)

						self.distributed_retriever_check(
						    self.get_dummy_ray_distributed_retriever(init_retrieval=lowerCAmelCase_	)     ,lowerCAmelCase_     ,lowerCAmelCase_	)
						ray.shutdown()







					@require_ray
					def  __lowerCAmelCase				( self    :       Optional[Any]	):
						lowerCAmelCase__			:		Any						       =			1
						lowerCAmelCase__			:		Union[str, Any]						       =			np.array(
						    [np.ones(self.retrieval_vector_size	), -np.ones(self.retrieval_vector_size	)]     ,dtype=np.floataa	)
						with self.assertRaises(lowerCAmelCase_	):
							self.distributed_retriever_check(
							    self.get_dummy_custom_hf_index_ray_retriever(init_retrieval=lowerCAmelCase_     ,from_disk=lowerCAmelCase_	)     ,lowerCAmelCase_     ,lowerCAmelCase_     ,)
						ray.shutdown()







					@require_ray
					def  __lowerCAmelCase				( self    :       str	):
						lowerCAmelCase__			:		List[Any]						       =			1
						lowerCAmelCase__			:		Optional[int]						       =			np.array(
						    [np.ones(self.retrieval_vector_size	), -np.ones(self.retrieval_vector_size	)]     ,dtype=np.floataa	)

						self.distributed_retriever_check(
						    self.get_dummy_custom_hf_index_ray_retriever(init_retrieval=lowerCAmelCase_     ,from_disk=lowerCAmelCase_	)     ,lowerCAmelCase_     ,lowerCAmelCase_	)
						ray.shutdown()







"""simple docstring"""





from typing import Dict, List, Optional, Union

import numpy as np

from ...image_processing_utils import BaseImageProcessor, BatchFeature, get_size_dict
from ...image_transforms import normalize, rescale, resize, to_channel_dimension_format
from ...image_utils import (
    IMAGENET_STANDARD_MEAN,
    IMAGENET_STANDARD_STD,
    ChannelDimension,
    ImageInput,
    PILImageResampling,
    make_list_of_images,
    to_numpy_array,
    valid_images,
)
from ...utils import TensorType, logging


__UpperCamelCase						:   List[Any]							   =	logging.get_logger(__name__)







class        SCREAMING_SNAKE_CASE      (	a_ ):






					"""simple docstring"""



					lowercase__       				= ["pixel_values"]







					def __init__( self    :       Tuple     ,lowercase_    :       bool = True     ,lowercase_    :       Optional[Dict[str, int]] = None     ,lowercase_    :       PILImageResampling = PILImageResampling.BILINEAR     ,lowercase_    :       bool = True     ,lowercase_    :       Union[int, float] = 1 / 2_5_5     ,lowercase_    :       bool = True     ,lowercase_    :       Optional[Union[float, List[float]]] = None     ,lowercase_    :       Optional[Union[float, List[float]]] = None     ,**lowercase_    :       Tuple     ,):
						super().__init__(**lowercase_	)
						lowerCAmelCase__			:		Any						       =			size if size is not None else {'''height''': 2_2_4, '''width''': 2_2_4}
						lowerCAmelCase__			:		Tuple						       =			get_size_dict(lowercase_	)
						lowerCAmelCase__			:		Tuple						       =			do_resize
						lowerCAmelCase__			:		Tuple						       =			do_rescale
						lowerCAmelCase__			:		Union[str, Any]						       =			do_normalize
						lowerCAmelCase__			:		Optional[int]						       =			size
						lowerCAmelCase__			:		List[Any]						       =			resample
						lowerCAmelCase__			:		int						       =			rescale_factor
						lowerCAmelCase__			:		int						       =			image_mean if image_mean is not None else IMAGENET_STANDARD_MEAN
						lowerCAmelCase__			:		Any						       =			image_std if image_std is not None else IMAGENET_STANDARD_STD







					def  __lowerCAmelCase				( self    :       List[Any]     ,lowercase_    :       np.ndarray     ,lowercase_    :       Dict[str, int]     ,lowercase_    :       PILImageResampling = PILImageResampling.BILINEAR     ,lowercase_    :       Optional[Union[str, ChannelDimension]] = None     ,**lowercase_    :       str     ,):
						lowerCAmelCase__			:		Union[str, Any]						       =			get_size_dict(lowercase_	)
						if "height" not in size or "width" not in size:
							raise ValueError(F'The `size` dictionary must contain the keys `height` and `width`. Got {size.keys()}'	)
						return resize(
						    lowercase_     ,size=(size['''height'''], size['''width'''])     ,resample=lowercase_     ,data_format=lowercase_     ,**lowercase_	)







					def  __lowerCAmelCase				( self    :       List[Any]     ,lowercase_    :       np.ndarray     ,lowercase_    :       float     ,lowercase_    :       Optional[Union[str, ChannelDimension]] = None     ,**lowercase_    :       Tuple	):
						return rescale(lowercase_     ,scale=lowercase_     ,data_format=lowercase_     ,**lowercase_	)







					def  __lowerCAmelCase				( self    :       Dict     ,lowercase_    :       np.ndarray     ,lowercase_    :       Union[float, List[float]]     ,lowercase_    :       Union[float, List[float]]     ,lowercase_    :       Optional[Union[str, ChannelDimension]] = None     ,**lowercase_    :       Union[str, Any]     ,):
						return normalize(lowercase_     ,mean=lowercase_     ,std=lowercase_     ,data_format=lowercase_     ,**lowercase_	)







					def  __lowerCAmelCase				( self    :       List[str]     ,lowercase_    :       ImageInput     ,lowercase_    :       Optional[bool] = None     ,lowercase_    :       Dict[str, int] = None     ,lowercase_    :       PILImageResampling = None     ,lowercase_    :       Optional[bool] = None     ,lowercase_    :       Optional[float] = None     ,lowercase_    :       Optional[bool] = None     ,lowercase_    :       Optional[Union[float, List[float]]] = None     ,lowercase_    :       Optional[Union[float, List[float]]] = None     ,lowercase_    :       Optional[Union[str, TensorType]] = None     ,lowercase_    :       Union[str, ChannelDimension] = ChannelDimension.FIRST     ,**lowercase_    :       List[Any]     ,):
						lowerCAmelCase__			:		Optional[int]						       =			do_resize if do_resize is not None else self.do_resize
						lowerCAmelCase__			:		Any						       =			do_rescale if do_rescale is not None else self.do_rescale
						lowerCAmelCase__			:		Dict						       =			do_normalize if do_normalize is not None else self.do_normalize
						lowerCAmelCase__			:		Dict						       =			resample if resample is not None else self.resample
						lowerCAmelCase__			:		Dict						       =			rescale_factor if rescale_factor is not None else self.rescale_factor
						lowerCAmelCase__			:		Optional[Any]						       =			image_mean if image_mean is not None else self.image_mean
						lowerCAmelCase__			:		str						       =			image_std if image_std is not None else self.image_std

						lowerCAmelCase__			:		Any						       =			size if size is not None else self.size
						lowerCAmelCase__			:		Dict						       =			get_size_dict(lowercase_	)

						lowerCAmelCase__			:		Union[str, Any]						       =			make_list_of_images(lowercase_	)

						if not valid_images(lowercase_	):
							raise ValueError(
							    '''Invalid image type. Must be of type PIL.Image.Image, numpy.ndarray, '''
							    '''torch.Tensor, tf.Tensor or jax.ndarray.'''	)

						if do_resize and size is None:
							raise ValueError('''Size must be specified if do_resize is True.'''	)

						if do_rescale and rescale_factor is None:
							raise ValueError('''Rescale factor must be specified if do_rescale is True.'''	)

						# All transformations expect numpy arrays.
						lowerCAmelCase__			:		Dict						       =			[to_numpy_array(lowercase_	) for image in images]

						if do_resize:
							lowerCAmelCase__			:		Optional[Any]						       =			[self.resize(image=lowercase_     ,size=lowercase_     ,resample=lowercase_	) for image in images]

						if do_rescale:
							lowerCAmelCase__			:		List[Any]						       =			[self.rescale(image=lowercase_     ,scale=lowercase_	) for image in images]

						if do_normalize:
							lowerCAmelCase__			:		Dict						       =			[self.normalize(image=lowercase_     ,mean=lowercase_     ,std=lowercase_	) for image in images]

						lowerCAmelCase__			:		Tuple						       =			[to_channel_dimension_format(lowercase_     ,lowercase_	) for image in images]

						lowerCAmelCase__			:		int						       =			{'''pixel_values''': images}
						return BatchFeature(data=lowercase_     ,tensor_type=lowercase_	)




