
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:
 lowerCamelCase_				:							Dict       =    None

if is_ray_available():
 import ray  # noqa: E402 # isort:skip
 from distributed_ray_retriever import RagRayDistributedRetriever, RayRetriever  # noqa: E402 # isort:skip
else:
 lowerCamelCase_				:							str       =    None
 lowerCamelCase_				:							Tuple       =    None
 lowerCamelCase_				:							Dict       =    None
def        __lowercase(							__snake_case				:		List[Any]					)       ->   Tuple:
       if not (is_datasets_available() and is_faiss_available() and is_psutil_available()):
              __snake_case						       =   unittest.skip('test requires Datasets, Faiss, psutil'					)(__snake_case					)
       return test_case







@require_distributed_retrieval
class  _lowerCamelCase			(lowerCamelCase					):

     def 						__lowerCamelCase				(  self    ):
            __snake_case						       =   tempfile.mkdtemp()
            __snake_case						       =   8

            # DPR tok
            __snake_case						       =   [
                '[UNK]',
                '[CLS]',
                '[SEP]',
                '[PAD]',
                '[MASK]',
                'want',
                '##want',
                '##ed',
                'wa',
                'un',
                'runn',
                '##ing',
                ',',
                'low',
                'lowest',
            ]
            __snake_case						       =   os.path.join(self.tmpdirname							,       'dpr_tokenizer'    )
            os.makedirs(SCREAMING_SNAKE_CASE_							,       exist_ok=SCREAMING_SNAKE_CASE_    )
            __snake_case						       =   os.path.join(SCREAMING_SNAKE_CASE_							,       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
            __snake_case						       =   [
                'l',
                'o',
                'w',
                'e',
                'r',
                's',
                't',
                'i',
                'd',
                'n',
                '\u0120',
                '\u0120l',
                '\u0120n',
                '\u0120lo',
                '\u0120low',
                'er',
                '\u0120lowest',
                '\u0120newer',
                '\u0120wider',
                '<unk>',
            ]
            __snake_case						       =   dict(zip(SCREAMING_SNAKE_CASE_							,       range(len(SCREAMING_SNAKE_CASE_    )    )    )    )
            __snake_case						       =   ['#version: 0.2', '\u0120 l', '\u0120l o', '\u0120lo w', 'e r', '']
            __snake_case						       =   {'unk_token': '<unk>'}

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

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

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

     def 						__lowerCamelCase				(  self    ):
            shutil.rmtree(self.tmpdirname    )

     def 						__lowerCamelCase				(  self    ):
            __snake_case						       =   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							,       SCREAMING_SNAKE_CASE_							,       SCREAMING_SNAKE_CASE_=12_345    ):
            __snake_case						       =   self.get_dummy_dataset()
            __snake_case						       =   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:
                   __snake_case						       =   dataset
                   __snake_case						       =   RagPyTorchDistributedRetriever(
                       SCREAMING_SNAKE_CASE_							,       question_encoder_tokenizer=self.get_dpr_tokenizer()							,       generator_tokenizer=self.get_bart_tokenizer()							,       )
                   if init_retrieval:
                          retriever.init_retrieval(SCREAMING_SNAKE_CASE_    )
            return retriever

     def 						__lowerCamelCase				(  self							,       SCREAMING_SNAKE_CASE_    ):
            # Have to run in local mode because sys.path modifications at top of
            # file are not propogated to remote workers.
            # https://stackoverflow.com/questions/54338013/parallel-import-a-python-file-from-sibling-folder
            ray.init(local_mode=SCREAMING_SNAKE_CASE_    )
            __snake_case						       =   RagConfig(
                retrieval_vector_size=self.retrieval_vector_size							,       question_encoder=DPRConfig().to_dict()							,       generator=BartConfig().to_dict()							,       )
            __snake_case						       =   ray.remote(SCREAMING_SNAKE_CASE_    )
            __snake_case						       =   [remote_cls.remote() for _ in range(1    )]
            with patch('transformers.models.rag.retrieval_rag.load_dataset'    ) as mock_load_dataset:
                   __snake_case						       =   self.get_dummy_dataset()
                   __snake_case						       =   RagRayDistributedRetriever(
                       SCREAMING_SNAKE_CASE_							,       question_encoder_tokenizer=self.get_dpr_tokenizer()							,       generator_tokenizer=self.get_bart_tokenizer()							,       retrieval_workers=SCREAMING_SNAKE_CASE_							,       )
                   if init_retrieval:
                          retriever.init_retrieval()
            return retriever

     def 						__lowerCamelCase				(  self							,       SCREAMING_SNAKE_CASE_							,       SCREAMING_SNAKE_CASE_							,       SCREAMING_SNAKE_CASE_=12_345    ):
            __snake_case						       =   self.get_dummy_dataset()
            __snake_case						       =   RagConfig(
                retrieval_vector_size=self.retrieval_vector_size							,       question_encoder=DPRConfig().to_dict()							,       generator=BartConfig().to_dict()							,       index_name='custom'							,       )
            if from_disk:
                   __snake_case						       =   os.path.join(self.tmpdirname							,       'dataset'    )
                   __snake_case						       =   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
                   __snake_case						       =   RagPyTorchDistributedRetriever(
                       SCREAMING_SNAKE_CASE_							,       question_encoder_tokenizer=self.get_dpr_tokenizer()							,       generator_tokenizer=self.get_bart_tokenizer()							,       )
            else:
                   __snake_case						       =   RagPyTorchDistributedRetriever(
                       SCREAMING_SNAKE_CASE_							,       question_encoder_tokenizer=self.get_dpr_tokenizer()							,       generator_tokenizer=self.get_bart_tokenizer()							,       index=CustomHFIndex(config.retrieval_vector_size							,       SCREAMING_SNAKE_CASE_    )							,       )
            if init_retrieval:
                   retriever.init_retrieval(SCREAMING_SNAKE_CASE_    )
            return retriever

     def 						__lowerCamelCase				(  self							,       SCREAMING_SNAKE_CASE_							,       SCREAMING_SNAKE_CASE_    ):
            # Have to run in local mode because sys.path modifications at top of
            # file are not propogated to remote workers.
            # https://stackoverflow.com/questions/54338013/parallel-import-a-python-file-from-sibling-folder
            ray.init(local_mode=SCREAMING_SNAKE_CASE_    )
            __snake_case						       =   self.get_dummy_dataset()
            __snake_case						       =   RagConfig(
                retrieval_vector_size=self.retrieval_vector_size							,       question_encoder=DPRConfig().to_dict()							,       generator=BartConfig().to_dict()							,       index_name='custom'							,       )
            __snake_case						       =   ray.remote(SCREAMING_SNAKE_CASE_    )
            __snake_case						       =   [remote_cls.remote() for _ in range(1    )]
            if from_disk:
                   __snake_case						       =   os.path.join(self.tmpdirname							,       'dataset'    )
                   __snake_case						       =   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
                   __snake_case						       =   RagRayDistributedRetriever(
                       SCREAMING_SNAKE_CASE_							,       question_encoder_tokenizer=self.get_dpr_tokenizer()							,       generator_tokenizer=self.get_bart_tokenizer()							,       retrieval_workers=SCREAMING_SNAKE_CASE_							,       index=CustomHFIndex.load_from_disk(
                           vector_size=config.retrieval_vector_size							,       dataset_path=config.passages_path							,       index_path=config.index_path							,       )							,       )
            else:
                   __snake_case						       =   RagRayDistributedRetriever(
                       SCREAMING_SNAKE_CASE_							,       question_encoder_tokenizer=self.get_dpr_tokenizer()							,       generator_tokenizer=self.get_bart_tokenizer()							,       retrieval_workers=SCREAMING_SNAKE_CASE_							,       index=CustomHFIndex(config.retrieval_vector_size							,       SCREAMING_SNAKE_CASE_    )							,       )
            if init_retrieval:
                   retriever.init_retrieval()
            return retriever

     def 						__lowerCamelCase				(  self							,       SCREAMING_SNAKE_CASE_							,       SCREAMING_SNAKE_CASE_							,       SCREAMING_SNAKE_CASE_    ):
            __snake_case       ,       __snake_case       ,       __snake_case						       =   retriever.retrieve(SCREAMING_SNAKE_CASE_							,       n_docs=SCREAMING_SNAKE_CASE_    )
            self.assertEqual(retrieved_doc_embeds.shape							,       (2, n_docs, self.retrieval_vector_size)    )
            self.assertEqual(len(SCREAMING_SNAKE_CASE_    )							,       2    )
            self.assertEqual(sorted(doc_dicts[0]    )							,       ['embeddings', 'id', 'text', 'title']    )
            self.assertEqual(len(doc_dicts[0]['id']    )							,       SCREAMING_SNAKE_CASE_    )
            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    ):
            __snake_case						       =   1
            __snake_case						       =   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=SCREAMING_SNAKE_CASE_    )							,       SCREAMING_SNAKE_CASE_							,       SCREAMING_SNAKE_CASE_    )

     def 						__lowerCamelCase				(  self    ):
            __snake_case						       =   1
            __snake_case						       =   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=SCREAMING_SNAKE_CASE_							,       from_disk=SCREAMING_SNAKE_CASE_    )							,       SCREAMING_SNAKE_CASE_							,       SCREAMING_SNAKE_CASE_							,       )

     def 						__lowerCamelCase				(  self    ):
            __snake_case						       =   1
            __snake_case						       =   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=SCREAMING_SNAKE_CASE_							,       from_disk=SCREAMING_SNAKE_CASE_    )							,       SCREAMING_SNAKE_CASE_							,       SCREAMING_SNAKE_CASE_							,       )

     @require_ray
     def 						__lowerCamelCase				(  self    ):
            __snake_case						       =   1
            __snake_case						       =   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=SCREAMING_SNAKE_CASE_    )							,       SCREAMING_SNAKE_CASE_							,       SCREAMING_SNAKE_CASE_    )
            ray.shutdown()

     @require_ray
     def 						__lowerCamelCase				(  self    ):
            __snake_case						       =   1
            __snake_case						       =   np.array(
                [np.ones(self.retrieval_vector_size    ), -np.ones(self.retrieval_vector_size    )]							,       dtype=np.floataa    )
            with self.assertRaises(SCREAMING_SNAKE_CASE_    ):
                   self.distributed_retriever_check(
                       self.get_dummy_custom_hf_index_ray_retriever(init_retrieval=SCREAMING_SNAKE_CASE_							,       from_disk=SCREAMING_SNAKE_CASE_    )							,       SCREAMING_SNAKE_CASE_							,       SCREAMING_SNAKE_CASE_							,       )
            ray.shutdown()

     @require_ray
     def 						__lowerCamelCase				(  self    ):
            __snake_case						       =   1
            __snake_case						       =   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=SCREAMING_SNAKE_CASE_							,       from_disk=SCREAMING_SNAKE_CASE_    )							,       SCREAMING_SNAKE_CASE_							,       SCREAMING_SNAKE_CASE_    )
            ray.shutdown()


import warnings
from dataclasses import dataclass, field
from typing import Optional, Tuple

from .training_args import TrainingArguments
from .utils import cached_property, is_tf_available, logging, requires_backends


lowerCamelCase_				:							Optional[Any]       =    logging.get_logger(__name__)

if is_tf_available():
 import tensorflow as tf







@dataclass
class  _lowerCamelCase			(lowerCamelCase					):
     lowercase__				  =			"""tf"""
     lowercase__				  =			field(
         default=lowerCamelCase							,					metadata={"""help""": """Name of TPU"""}							,					)

     lowercase__				  =			field(
         default=lowerCamelCase							,					metadata={"""help""": """Zone of TPU"""}							,					)

     lowercase__				  =			field(
         default=lowerCamelCase							,					metadata={"""help""": """Name of Cloud TPU-enabled project"""}							,					)

     lowercase__				  =			field(
         default=1.0							,					metadata={"""help""": """Power for the Polynomial decay LR scheduler."""}							,					)

     lowercase__				  =			field(default=lowerCamelCase							,					metadata={"""help""": """Whether to activate the XLA compilation or not"""}					)

     @cached_property
     def 						__lowerCamelCase				(  self    ):
            requires_backends(self							,       ['tf']    )
            logger.info('Tensorflow: setting up strategy'    )

            __snake_case						       =   tf.config.list_physical_devices('GPU'    )

            # Set to float16 at first
            if self.fpaa:
                   tf.keras.mixed_precision.set_global_policy('mixed_float16'    )

            if self.no_cuda:
                   __snake_case						       =   tf.distribute.OneDeviceStrategy(device='/cpu:0'    )
            else:
                   try:
                          if self.tpu_name:
                                 __snake_case						       =   tf.distribute.cluster_resolver.TPUClusterResolver(
                                     self.tpu_name							,       zone=self.tpu_zone							,       project=self.gcp_project    )
                          else:
                                 __snake_case						       =   tf.distribute.cluster_resolver.TPUClusterResolver()
                   except ValueError:
                          if self.tpu_name:
                                 raise RuntimeError(f'''Couldn\'t connect to TPU {self.tpu_name}!'''    )
                          else:
                                 __snake_case						       =   None

                   if tpu:
                          # Set to bfloat16 in case of TPU
                          if self.fpaa:
                                 tf.keras.mixed_precision.set_global_policy('mixed_bfloat16'    )

                          tf.config.experimental_connect_to_cluster(SCREAMING_SNAKE_CASE_    )
                          tf.tpu.experimental.initialize_tpu_system(SCREAMING_SNAKE_CASE_    )

                          __snake_case						       =   tf.distribute.TPUStrategy(SCREAMING_SNAKE_CASE_    )

                   elif len(SCREAMING_SNAKE_CASE_    ) == 0:
                          __snake_case						       =   tf.distribute.OneDeviceStrategy(device='/cpu:0'    )
                   elif len(SCREAMING_SNAKE_CASE_    ) == 1:
                          __snake_case						       =   tf.distribute.OneDeviceStrategy(device='/gpu:0'    )
                   elif len(SCREAMING_SNAKE_CASE_    ) > 1:
                          # If you only want to use a specific subset of GPUs use `CUDA_VISIBLE_DEVICES=0`
                          __snake_case						       =   tf.distribute.MirroredStrategy()
                   else:
                          raise ValueError('Cannot find the proper strategy, please check your environment properties.'    )

            return strategy

     @property
     def 						__lowerCamelCase				(  self    ):
            requires_backends(self							,       ['tf']    )
            return self._setup_strategy

     @property
     def 						__lowerCamelCase				(  self    ):
            requires_backends(self							,       ['tf']    )
            return self._setup_strategy.num_replicas_in_sync

     @property
     def 						__lowerCamelCase				(  self    ):
            return False  # TF Logging is handled by Keras not the Trainer

     @property
     def 						__lowerCamelCase				(  self    ):
            if self.per_gpu_train_batch_size:
                   logger.warning(
                       'Using deprecated `--per_gpu_train_batch_size` argument which will be removed in a future '
                       'version. Using `--per_device_train_batch_size` is preferred.'    )
            __snake_case						       =   self.per_gpu_train_batch_size or self.per_device_train_batch_size
            return per_device_batch_size * self.n_replicas

     @property
     def 						__lowerCamelCase				(  self    ):
            if self.per_gpu_eval_batch_size:
                   logger.warning(
                       'Using deprecated `--per_gpu_eval_batch_size` argument which will be removed in a future '
                       'version. Using `--per_device_eval_batch_size` is preferred.'    )
            __snake_case						       =   self.per_gpu_eval_batch_size or self.per_device_eval_batch_size
            return per_device_batch_size * self.n_replicas

     @property
     def 						__lowerCamelCase				(  self    ):
            requires_backends(self							,       ['tf']    )
            warnings.warn(
                'The n_gpu argument is deprecated and will be removed in a future version, use n_replicas instead.'							,       SCREAMING_SNAKE_CASE_							,       )
            return self._setup_strategy.num_replicas_in_sync
