


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:
    A				:   Tuple		= None

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

def 				lowercase_		(		_A      :   List[str]						):

      """simple docstring"""




      if not (is_datasets_available() and is_faiss_available() and is_psutil_available()):
            lowerCamelCase__		:    List[Any]   =  unittest.skip("test requires Datasets, Faiss, psutil"						)(snake_case__						)
      return test_case







@require_distributed_retrieval
class       _lowercase			(							SCREAMING_SNAKE_CASE__):


       """simple docstring"""




       def 						lowerCAmelCase					(       self							:      Tuple	):


             '''simple docstring'''



             lowerCamelCase__		:    Optional[int]   =  tempfile.mkdtemp()
             lowerCamelCase__		:    Optional[int]   =  8

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

             lowerCamelCase__		:    Union[str, Any]   =  os.path.join(self.tmpdirname ,				"bart_tokenizer"	)
             os.makedirs(_UpperCAmelCase ,				exist_ok=_UpperCAmelCase	)
             lowerCamelCase__		:    Optional[int]   =  os.path.join(_UpperCAmelCase ,				BART_VOCAB_FILES_NAMES["vocab_file"]	)
             lowerCamelCase__		:    Dict   =  os.path.join(_UpperCAmelCase ,				BART_VOCAB_FILES_NAMES["merges_file"]	)
             with open(self.vocab_file ,				"w" ,				encoding="utf-8"	) as fp:
                   fp.write(json.dumps(_UpperCAmelCase	) + "\n"	)
             with open(self.merges_file ,				"w" ,				encoding="utf-8"	) as fp:
                   fp.write("\n".join(_UpperCAmelCase	)	)





       def 						lowerCAmelCase					(       self							:      List[str]	):


             '''simple docstring'''



             return DPRQuestionEncoderTokenizer.from_pretrained(os.path.join(self.tmpdirname ,				"dpr_tokenizer"	)	)





       def 						lowerCAmelCase					(       self							:      List[str]	):


             '''simple docstring'''



             return BartTokenizer.from_pretrained(os.path.join(self.tmpdirname ,				"bart_tokenizer"	)	)





       def 						lowerCAmelCase					(       self							:      Union[str, Any]	):


             '''simple docstring'''



             shutil.rmtree(self.tmpdirname	)





       def 						lowerCAmelCase					(       self							:      Optional[int]	):


             '''simple docstring'''



             lowerCamelCase__		:    List[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							:      int ,				__lowerCamelCase							:      Optional[Any] ,				__lowerCamelCase							:      Union[str, Any]=12345	):


             '''simple docstring'''



             lowerCamelCase__		:    Union[str, Any]   =  self.get_dummy_dataset()
             lowerCamelCase__		:    Optional[int]   =  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[Any]   =  dataset
                   lowerCamelCase__		:    List[str]   =  RagPyTorchDistributedRetriever(
                       _UpperCAmelCase ,				question_encoder_tokenizer=self.get_dpr_tokenizer() ,				generator_tokenizer=self.get_bart_tokenizer() ,				)
                   if init_retrieval:
                         retriever.init_retrieval(_UpperCAmelCase	)
             return retriever





       def 						lowerCAmelCase					(       self							:      Union[str, Any] ,				__lowerCamelCase							:      int	):


             '''simple docstring'''



             ray.init(local_mode=_UpperCAmelCase	)
             lowerCamelCase__		:    str   =  RagConfig(
                 retrieval_vector_size=self.retrieval_vector_size ,				question_encoder=DPRConfig().to_dict() ,				generator=BartConfig().to_dict() ,				)
             lowerCamelCase__		:    str   =  ray.remote(_UpperCAmelCase	)
             lowerCamelCase__		:    str   =  [remote_cls.remote() for _ in range(1	)]
             with patch("transformers.models.rag.retrieval_rag.load_dataset"	) as mock_load_dataset:
                   lowerCamelCase__		:    str   =  self.get_dummy_dataset()
                   lowerCamelCase__		:    Union[str, Any]   =  RagRayDistributedRetriever(
                       _UpperCAmelCase ,				question_encoder_tokenizer=self.get_dpr_tokenizer() ,				generator_tokenizer=self.get_bart_tokenizer() ,				retrieval_workers=_UpperCAmelCase ,				)
                   if init_retrieval:
                         retriever.init_retrieval()
             return retriever





       def 						lowerCAmelCase					(       self							:      int ,				__lowerCamelCase							:      Union[str, Any] ,				__lowerCamelCase							:      Optional[int] ,				__lowerCamelCase							:      List[str]=12345	):


             '''simple docstring'''



             lowerCamelCase__		:    Union[str, Any]   =  self.get_dummy_dataset()
             lowerCamelCase__		:    Optional[int]   =  RagConfig(
                 retrieval_vector_size=self.retrieval_vector_size ,				question_encoder=DPRConfig().to_dict() ,				generator=BartConfig().to_dict() ,				index_name="custom" ,				)
             if from_disk:
                   lowerCamelCase__		:    Optional[int]   =  os.path.join(self.tmpdirname ,				"dataset"	)
                   lowerCamelCase__		:    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__		:    List[str]   =  RagPyTorchDistributedRetriever(
                       _UpperCAmelCase ,				question_encoder_tokenizer=self.get_dpr_tokenizer() ,				generator_tokenizer=self.get_bart_tokenizer() ,				)
             else:
                   lowerCamelCase__		:    Any   =  RagPyTorchDistributedRetriever(
                       _UpperCAmelCase ,				question_encoder_tokenizer=self.get_dpr_tokenizer() ,				generator_tokenizer=self.get_bart_tokenizer() ,				index=CustomHFIndex(config.retrieval_vector_size ,				_UpperCAmelCase	) ,				)
             if init_retrieval:
                   retriever.init_retrieval(_UpperCAmelCase	)
             return retriever





       def 						lowerCAmelCase					(       self							:      Union[str, Any] ,				__lowerCamelCase							:      Tuple ,				__lowerCamelCase							:      List[str]	):


             '''simple docstring'''



             ray.init(local_mode=_UpperCAmelCase	)
             lowerCamelCase__		:    Optional[int]   =  self.get_dummy_dataset()
             lowerCamelCase__		:    Dict   =  RagConfig(
                 retrieval_vector_size=self.retrieval_vector_size ,				question_encoder=DPRConfig().to_dict() ,				generator=BartConfig().to_dict() ,				index_name="custom" ,				)
             lowerCamelCase__		:    Union[str, Any]   =  ray.remote(_UpperCAmelCase	)
             lowerCamelCase__		:    str   =  [remote_cls.remote() for _ in range(1	)]
             if from_disk:
                   lowerCamelCase__		:    Tuple   =  os.path.join(self.tmpdirname ,				"dataset"	)
                   lowerCamelCase__		:    List[Any]   =  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__		:    Tuple   =  RagRayDistributedRetriever(
                       _UpperCAmelCase ,				question_encoder_tokenizer=self.get_dpr_tokenizer() ,				generator_tokenizer=self.get_bart_tokenizer() ,				retrieval_workers=_UpperCAmelCase ,				index=CustomHFIndex.load_from_disk(
                           vector_size=config.retrieval_vector_size ,				dataset_path=config.passages_path ,				index_path=config.index_path ,				) ,				)
             else:
                   lowerCamelCase__		:    List[Any]   =  RagRayDistributedRetriever(
                       _UpperCAmelCase ,				question_encoder_tokenizer=self.get_dpr_tokenizer() ,				generator_tokenizer=self.get_bart_tokenizer() ,				retrieval_workers=_UpperCAmelCase ,				index=CustomHFIndex(config.retrieval_vector_size ,				_UpperCAmelCase	) ,				)
             if init_retrieval:
                   retriever.init_retrieval()
             return retriever





       def 						lowerCAmelCase					(       self							:      str ,				__lowerCamelCase							:      int ,				__lowerCamelCase							:      Dict ,				__lowerCamelCase							:      str	):


             '''simple docstring'''



             lowerCamelCase__		:    Union[str, Any]   =  retriever.retrieve(_UpperCAmelCase ,				n_docs=_UpperCAmelCase	)
             self.assertEqual(retrieved_doc_embeds.shape ,				(2, n_docs, self.retrieval_vector_size)	)
             self.assertEqual(len(_UpperCAmelCase	) ,				2	)
             self.assertEqual(sorted(doc_dicts[0]	) ,				["embeddings", "id", "text", "title"]	)
             self.assertEqual(len(doc_dicts[0]["id"]	) ,				_UpperCAmelCase	)
             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							:      str	):


             '''simple docstring'''



             lowerCamelCase__		:    Optional[Any]   =  1
             lowerCamelCase__		:    Optional[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_pytorch_distributed_retriever(init_retrieval=_UpperCAmelCase	) ,				_UpperCAmelCase ,				_UpperCAmelCase	)





       def 						lowerCAmelCase					(       self							:      Optional[Any]	):


             '''simple docstring'''



             lowerCamelCase__		:    Optional[int]   =  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_custom_hf_index_pytorch_retriever(init_retrieval=_UpperCAmelCase ,				from_disk=_UpperCAmelCase	) ,				_UpperCAmelCase ,				_UpperCAmelCase ,				)





       def 						lowerCAmelCase					(       self							:      Optional[Any]	):


             '''simple docstring'''



             lowerCamelCase__		:    Optional[int]   =  1
             lowerCamelCase__		:    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=_UpperCAmelCase ,				from_disk=_UpperCAmelCase	) ,				_UpperCAmelCase ,				_UpperCAmelCase ,				)





       @require_ray
       def 						lowerCAmelCase					(       self							:      Optional[Any]	):


             '''simple docstring'''



             lowerCamelCase__		:    List[str]   =  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_ray_distributed_retriever(init_retrieval=_UpperCAmelCase	) ,				_UpperCAmelCase ,				_UpperCAmelCase	)
             ray.shutdown()





       @require_ray
       def 						lowerCAmelCase					(       self							:      Tuple	):


             '''simple docstring'''



             lowerCamelCase__		:    Optional[int]   =  1
             lowerCamelCase__		:    Any   =  np.array(
                 [np.ones(self.retrieval_vector_size	), -np.ones(self.retrieval_vector_size	)] ,				dtype=np.floataa	)
             with self.assertRaises(_UpperCAmelCase	):
                   self.distributed_retriever_check(
                       self.get_dummy_custom_hf_index_ray_retriever(init_retrieval=_UpperCAmelCase ,				from_disk=_UpperCAmelCase	) ,				_UpperCAmelCase ,				_UpperCAmelCase ,				)
             ray.shutdown()





       @require_ray
       def 						lowerCAmelCase					(       self							:      int	):


             '''simple docstring'''



             lowerCamelCase__		:    str   =  1
             lowerCamelCase__		:    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_ray_retriever(init_retrieval=_UpperCAmelCase ,				from_disk=_UpperCAmelCase	) ,				_UpperCAmelCase ,				_UpperCAmelCase	)
             ray.shutdown()






import warnings
from contextlib import contextmanager

from ...processing_utils import ProcessorMixin







class       _lowercase			(							lowercase__):


       """simple docstring"""




       A__ 						=      "AutoFeatureExtractor"
       A__ 						=      "Speech2Text2Tokenizer"





       def __init__(       self							:      List[str] ,				__lowerCamelCase							:      Dict ,				__lowerCamelCase							:      str	):


             '''simple docstring'''



             super().__init__(__lowerCamelCase ,				__lowerCamelCase	)
             lowerCamelCase__		:    Optional[Any]   =  self.feature_extractor
             lowerCamelCase__		:    Optional[int]   =  False





       def __call__(       self							:      List[str] ,				*__lowerCamelCase							:      Union[str, Any] ,				**__lowerCamelCase							:      List[Any]	):


             '''simple docstring'''



             if self._in_target_context_manager:
                   return self.current_processor(*__lowerCamelCase ,				**__lowerCamelCase	)

             if "raw_speech" in kwargs:
                   warnings.warn("Using `raw_speech` as a keyword argument is deprecated. Use `audio` instead."	)
                   lowerCamelCase__		:    List[Any]   =  kwargs.pop("raw_speech"	)
             else:
                   lowerCamelCase__		:    str   =  kwargs.pop("audio" ,				__lowerCamelCase	)
             lowerCamelCase__		:    Optional[Any]   =  kwargs.pop("sampling_rate" ,				__lowerCamelCase	)
             lowerCamelCase__		:    Union[str, Any]   =  kwargs.pop("text" ,				__lowerCamelCase	)
             if len(__lowerCamelCase	) > 0:
                   lowerCamelCase__		:    Dict   =  args[0]
                   lowerCamelCase__		:    int   =  args[1:]

             if audio is None and text is None:
                   raise ValueError("You need to specify either an `audio` or `text` input to process."	)

             if audio is not None:
                   lowerCamelCase__		:    str   =  self.feature_extractor(__lowerCamelCase ,				*__lowerCamelCase ,				sampling_rate=__lowerCamelCase ,				**__lowerCamelCase	)
             if text is not None:
                   lowerCamelCase__		:    Optional[int]   =  self.tokenizer(__lowerCamelCase ,				**__lowerCamelCase	)

             if text is None:
                   return inputs
             elif audio is None:
                   return encodings
             else:
                   lowerCamelCase__		:    Optional[int]   =  encodings["input_ids"]
                   return inputs





       def 						lowerCAmelCase					(       self							:      Any ,				*__lowerCamelCase							:      Any ,				**__lowerCamelCase							:      Any	):


             '''simple docstring'''



             return self.tokenizer.batch_decode(*__lowerCamelCase ,				**__lowerCamelCase	)





       def 						lowerCAmelCase					(       self							:      Dict ,				*__lowerCamelCase							:      int ,				**__lowerCamelCase							:      List[str]	):


             '''simple docstring'''



             return self.tokenizer.decode(*__lowerCamelCase ,				**__lowerCamelCase	)





       @contextmanager
       def 						lowerCAmelCase					(       self							:      Any	):


             '''simple docstring'''



             warnings.warn(
                 "`as_target_processor` is deprecated and will be removed in v5 of Transformers. You can process your "
                 "labels by using the argument `text` of the regular `__call__` method (either in the same call as "
                 "your audio inputs, or in a separate call."	)
             lowerCamelCase__		:    int   =  True
             lowerCamelCase__		:    List[Any]   =  self.tokenizer
             yield
             lowerCamelCase__		:    Optional[int]   =  self.feature_extractor
             lowerCamelCase__		:    List[str]   =  False


