

from typing import Dict, List, Optional, Union

import numpy as np

from transformers.utils import is_vision_available
from transformers.utils.generic import TensorType

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


if is_vision_available():
  import PIL

UpperCAmelCase__ :     str				    =    logging.get_logger(__name__)






def 					__lowercase      (     _A			)     ->						List[List[ImageInput]]:
      if isinstance(_A						,    (list, tuple)			) and isinstance(videos[0]						,    (list, tuple)			) and is_valid_image(videos[0][0]			):
            return videos

      elif isinstance(_A						,    (list, tuple)			) and is_valid_image(videos[0]			):
            return [videos]

      elif is_valid_image(_A			):
            return [[videos]]

      raise ValueError(F"Could not make batched video from {videos}"			)







class        a__       (   UpperCAmelCase       ):







  """simple docstring"""




  UpperCAmelCase__      : Dict			       =["""pixel_values"""]


  def __init__(			self    : Union[str, Any]				,				UpperCAmelCase__    : bool = True				,				UpperCAmelCase__    : Dict[str, int] = None				,				UpperCAmelCase__    : PILImageResampling = PILImageResampling.BILINEAR				,				UpperCAmelCase__    : bool = True				,				UpperCAmelCase__    : Dict[str, int] = None				,				UpperCAmelCase__    : bool = True				,				UpperCAmelCase__    : Union[int, float] = 1 / 2_5_5				,				UpperCAmelCase__    : bool = True				,				UpperCAmelCase__    : bool = True				,				UpperCAmelCase__    : Optional[Union[float, List[float]]] = None				,				UpperCAmelCase__    : Optional[Union[float, List[float]]] = None				,				**UpperCAmelCase__    : Tuple				,				)					->None:
        """simple docstring"""







        super().__init__(**UpperCAmelCase__		)
        SCREAMING_SNAKE_CASE :      Union[str, Any]       =   size if size is not None else {"""shortest_edge""": 2_5_6}
        SCREAMING_SNAKE_CASE :      List[Any]       =   get_size_dict(UpperCAmelCase__				,				default_to_square=UpperCAmelCase__		)
        SCREAMING_SNAKE_CASE :      Optional[Any]       =   crop_size if crop_size is not None else {"""height""": 2_2_4, """width""": 2_2_4}
        SCREAMING_SNAKE_CASE :      Union[str, Any]       =   get_size_dict(UpperCAmelCase__				,				param_name="""crop_size"""		)

        SCREAMING_SNAKE_CASE :      List[Any]       =   do_resize
        SCREAMING_SNAKE_CASE :      Tuple       =   size
        SCREAMING_SNAKE_CASE :      str       =   do_center_crop
        SCREAMING_SNAKE_CASE :      List[str]       =   crop_size
        SCREAMING_SNAKE_CASE :      List[Any]       =   resample
        SCREAMING_SNAKE_CASE :      Union[str, Any]       =   do_rescale
        SCREAMING_SNAKE_CASE :      Union[str, Any]       =   rescale_factor
        SCREAMING_SNAKE_CASE :      Dict       =   offset
        SCREAMING_SNAKE_CASE :      str       =   do_normalize
        SCREAMING_SNAKE_CASE :      Optional[int]       =   image_mean if image_mean is not None else IMAGENET_STANDARD_MEAN
        SCREAMING_SNAKE_CASE :      Tuple       =   image_std if image_std is not None else IMAGENET_STANDARD_STD


  def  _lowercase      (			self    : Dict				,				UpperCAmelCase__    : np.ndarray				,				UpperCAmelCase__    : Dict[str, int]				,				UpperCAmelCase__    : PILImageResampling = PILImageResampling.BILINEAR				,				UpperCAmelCase__    : Optional[Union[str, ChannelDimension]] = None				,				**UpperCAmelCase__    : Optional[Any]				,				)					->np.ndarray:
        """simple docstring"""







        SCREAMING_SNAKE_CASE :      Tuple       =   get_size_dict(UpperCAmelCase__				,				default_to_square=UpperCAmelCase__		)
        if "shortest_edge" in size:
              SCREAMING_SNAKE_CASE :      Optional[int]       =   get_resize_output_image_size(UpperCAmelCase__				,				size["""shortest_edge"""]				,				default_to_square=UpperCAmelCase__		)
        elif "height" in size and "width" in size:
              SCREAMING_SNAKE_CASE :      Any       =   (size["""height"""], size["""width"""])
        else:
              raise ValueError(f"Size must have 'height' and 'width' or 'shortest_edge' as keys. Got {size.keys()}"		)
        return resize(UpperCAmelCase__				,				size=UpperCAmelCase__				,				resample=UpperCAmelCase__				,				data_format=UpperCAmelCase__				,				**UpperCAmelCase__		)


  def  _lowercase      (			self    : List[Any]				,				UpperCAmelCase__    : np.ndarray				,				UpperCAmelCase__    : Dict[str, int]				,				UpperCAmelCase__    : Optional[Union[str, ChannelDimension]] = None				,				**UpperCAmelCase__    : List[str]				,				)					->np.ndarray:
        """simple docstring"""







        SCREAMING_SNAKE_CASE :      Union[str, Any]       =   get_size_dict(UpperCAmelCase__		)
        if "height" not in size or "width" not in size:
              raise ValueError(f"Size must have 'height' and 'width' as keys. Got {size.keys()}"		)
        return center_crop(UpperCAmelCase__				,				size=(size["""height"""], size["""width"""])				,				data_format=UpperCAmelCase__				,				**UpperCAmelCase__		)


  def  _lowercase      (			self    : List[str]				,				UpperCAmelCase__    : np.ndarray				,				UpperCAmelCase__    : Union[int, float]				,				UpperCAmelCase__    : bool = True				,				UpperCAmelCase__    : Optional[Union[str, ChannelDimension]] = None				,				**UpperCAmelCase__    : Optional[int]				,				)					->List[Any]:
        """simple docstring"""







        SCREAMING_SNAKE_CASE :      Optional[int]       =   image.astype(np.floataa		)
        if offset:
              SCREAMING_SNAKE_CASE :      int       =   image - (scale / 2)
        return rescale(UpperCAmelCase__				,				scale=UpperCAmelCase__				,				data_format=UpperCAmelCase__				,				**UpperCAmelCase__		)


  def  _lowercase      (			self    : Union[str, Any]				,				UpperCAmelCase__    : np.ndarray				,				UpperCAmelCase__    : Union[float, List[float]]				,				UpperCAmelCase__    : Union[float, List[float]]				,				UpperCAmelCase__    : Optional[Union[str, ChannelDimension]] = None				,				**UpperCAmelCase__    : str				,				)					->np.ndarray:
        """simple docstring"""







        return normalize(UpperCAmelCase__				,				mean=UpperCAmelCase__				,				std=UpperCAmelCase__				,				data_format=UpperCAmelCase__				,				**UpperCAmelCase__		)


  def  _lowercase      (			self    : Optional[int]				,				UpperCAmelCase__    : ImageInput				,				UpperCAmelCase__    : bool = None				,				UpperCAmelCase__    : Dict[str, int] = None				,				UpperCAmelCase__    : PILImageResampling = None				,				UpperCAmelCase__    : bool = None				,				UpperCAmelCase__    : Dict[str, int] = None				,				UpperCAmelCase__    : bool = None				,				UpperCAmelCase__    : float = None				,				UpperCAmelCase__    : bool = None				,				UpperCAmelCase__    : bool = None				,				UpperCAmelCase__    : Optional[Union[float, List[float]]] = None				,				UpperCAmelCase__    : Optional[Union[float, List[float]]] = None				,				UpperCAmelCase__    : Optional[ChannelDimension] = ChannelDimension.FIRST				,				)					->np.ndarray:
        """simple docstring"""







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

        if do_center_crop and crop_size is None:
              raise ValueError("""Crop size must be specified if do_center_crop is True."""		)

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

        if do_normalize and (image_mean is None or image_std is None):
              raise ValueError("""Image mean and std must be specified if do_normalize is True."""		)

        if offset and not do_rescale:
              raise ValueError("""For offset, do_rescale must also be set to True."""		)

        # All transformations expect numpy arrays.
        SCREAMING_SNAKE_CASE :      List[str]       =   to_numpy_array(UpperCAmelCase__		)

        if do_resize:
              SCREAMING_SNAKE_CASE :      Tuple       =   self.resize(image=UpperCAmelCase__				,				size=UpperCAmelCase__				,				resample=UpperCAmelCase__		)

        if do_center_crop:
              SCREAMING_SNAKE_CASE :      Optional[int]       =   self.center_crop(UpperCAmelCase__				,				size=UpperCAmelCase__		)

        if do_rescale:
              SCREAMING_SNAKE_CASE :      Dict       =   self.rescale(image=UpperCAmelCase__				,				scale=UpperCAmelCase__				,				offset=UpperCAmelCase__		)

        if do_normalize:
              SCREAMING_SNAKE_CASE :      Tuple       =   self.normalize(image=UpperCAmelCase__				,				mean=UpperCAmelCase__				,				std=UpperCAmelCase__		)

        SCREAMING_SNAKE_CASE :      Tuple       =   to_channel_dimension_format(UpperCAmelCase__				,				UpperCAmelCase__		)
        return image




  def  _lowercase      (			self    : Dict				,				UpperCAmelCase__    : ImageInput				,				UpperCAmelCase__    : bool = None				,				UpperCAmelCase__    : Dict[str, int] = None				,				UpperCAmelCase__    : PILImageResampling = None				,				UpperCAmelCase__    : bool = None				,				UpperCAmelCase__    : Dict[str, int] = None				,				UpperCAmelCase__    : bool = None				,				UpperCAmelCase__    : float = None				,				UpperCAmelCase__    : bool = None				,				UpperCAmelCase__    : bool = None				,				UpperCAmelCase__    : Optional[Union[float, List[float]]] = None				,				UpperCAmelCase__    : Optional[Union[float, List[float]]] = None				,				UpperCAmelCase__    : Optional[Union[str, TensorType]] = None				,				UpperCAmelCase__    : ChannelDimension = ChannelDimension.FIRST				,				**UpperCAmelCase__    : Dict				,				)					->PIL.Image.Image:
        """simple docstring"""







        SCREAMING_SNAKE_CASE :      Tuple       =   do_resize if do_resize is not None else self.do_resize
        SCREAMING_SNAKE_CASE :      str       =   resample if resample is not None else self.resample
        SCREAMING_SNAKE_CASE :      List[str]       =   do_center_crop if do_center_crop is not None else self.do_center_crop
        SCREAMING_SNAKE_CASE :      List[str]       =   do_rescale if do_rescale is not None else self.do_rescale
        SCREAMING_SNAKE_CASE :      List[str]       =   rescale_factor if rescale_factor is not None else self.rescale_factor
        SCREAMING_SNAKE_CASE :      List[Any]       =   offset if offset is not None else self.offset
        SCREAMING_SNAKE_CASE :      List[str]       =   do_normalize if do_normalize is not None else self.do_normalize
        SCREAMING_SNAKE_CASE :      Optional[int]       =   image_mean if image_mean is not None else self.image_mean
        SCREAMING_SNAKE_CASE :      Tuple       =   image_std if image_std is not None else self.image_std

        SCREAMING_SNAKE_CASE :      List[Any]       =   size if size is not None else self.size
        SCREAMING_SNAKE_CASE :      Optional[Any]       =   get_size_dict(UpperCAmelCase__				,				default_to_square=UpperCAmelCase__		)
        SCREAMING_SNAKE_CASE :      Any       =   crop_size if crop_size is not None else self.crop_size
        SCREAMING_SNAKE_CASE :      Union[str, Any]       =   get_size_dict(UpperCAmelCase__				,				param_name="""crop_size"""		)

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

        SCREAMING_SNAKE_CASE :      Dict       =   make_batched(UpperCAmelCase__		)

        SCREAMING_SNAKE_CASE :      Dict       =   [
            [
                self._preprocess_image(
                    image=UpperCAmelCase__				,				do_resize=UpperCAmelCase__				,				size=UpperCAmelCase__				,				resample=UpperCAmelCase__				,				do_center_crop=UpperCAmelCase__				,				crop_size=UpperCAmelCase__				,				do_rescale=UpperCAmelCase__				,				rescale_factor=UpperCAmelCase__				,				offset=UpperCAmelCase__				,				do_normalize=UpperCAmelCase__				,				image_mean=UpperCAmelCase__				,				image_std=UpperCAmelCase__				,				data_format=UpperCAmelCase__				,				)
                for img in video
            ]
            for video in videos
        ]

        SCREAMING_SNAKE_CASE :      Optional[int]       =   {"""pixel_values""": videos}
        return BatchFeature(data=UpperCAmelCase__				,				tensor_type=UpperCAmelCase__		)







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

import numpy as np
from datasets import Dataset

from transformers import is_faiss_available
from transformers.models.bart.configuration_bart import BartConfig
from transformers.models.bart.tokenization_bart import BartTokenizer
from transformers.models.bert.tokenization_bert import VOCAB_FILES_NAMES as DPR_VOCAB_FILES_NAMES
from transformers.models.dpr.configuration_dpr import DPRConfig
from transformers.models.dpr.tokenization_dpr import DPRContextEncoderTokenizer, DPRQuestionEncoderTokenizer
from transformers.models.rag.configuration_rag import RagConfig
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_faiss, require_sentencepiece, require_tokenizers, require_torch


if is_faiss_available():
  import faiss







@require_faiss
class        a__       (   UpperCAmelCase       ):







  """simple docstring"""




  def  _lowercase      (			self    : Optional[int]		)					->Optional[int]:
        """simple docstring"""







        SCREAMING_SNAKE_CASE :      List[Any]       =   tempfile.mkdtemp()
        SCREAMING_SNAKE_CASE :      Optional[Any]       =   8

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

        SCREAMING_SNAKE_CASE :      int       =   os.path.join(self.tmpdirname				,				"""bart_tokenizer"""		)
        os.makedirs(UpperCAmelCase__				,				exist_ok=UpperCAmelCase__		)
        SCREAMING_SNAKE_CASE :      Optional[Any]       =   os.path.join(UpperCAmelCase__				,				BART_VOCAB_FILES_NAMES["""vocab_file"""]		)
        SCREAMING_SNAKE_CASE :      Optional[int]       =   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  _lowercase      (			self    : Optional[Any]		)					->DPRQuestionEncoderTokenizer:
        """simple docstring"""







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


  def  _lowercase      (			self    : Optional[Any]		)					->DPRContextEncoderTokenizer:
        """simple docstring"""







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


  def  _lowercase      (			self    : Optional[int]		)					->BartTokenizer:
        """simple docstring"""







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


  def  _lowercase      (			self    : Any		)					->List[str]:
        """simple docstring"""







        shutil.rmtree(self.tmpdirname		)


  def  _lowercase      (			self    : Tuple		)					->Union[str, Any]:
        """simple docstring"""







        SCREAMING_SNAKE_CASE :      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  _lowercase      (			self    : Union[str, Any]		)					->Tuple:
        """simple docstring"""







        SCREAMING_SNAKE_CASE :      int       =   self.get_dummy_dataset()
        SCREAMING_SNAKE_CASE :      str       =   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:
              SCREAMING_SNAKE_CASE :      Tuple       =   dataset
              SCREAMING_SNAKE_CASE :      Union[str, Any]       =   RagRetriever(
                  UpperCAmelCase__				,				question_encoder_tokenizer=self.get_dpr_tokenizer()				,				generator_tokenizer=self.get_bart_tokenizer()				,				)
        return retriever


  def  _lowercase      (			self    : List[Any]				,				UpperCAmelCase__    : bool		)					->Union[str, Any]:
        """simple docstring"""







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


  def  _lowercase      (			self    : int		)					->int:
        """simple docstring"""







        SCREAMING_SNAKE_CASE :      Optional[Any]       =   Dataset.from_dict(
            {
                """id""": ["""0""", """1"""],
                """text""": ["""foo""", """bar"""],
                """title""": ["""Foo""", """Bar"""],
                """embeddings""": [np.ones(self.retrieval_vector_size + 1		), 2 * np.ones(self.retrieval_vector_size + 1		)],
            }		)
        dataset.add_faiss_index("""embeddings"""				,				string_factory="""Flat"""				,				metric_type=faiss.METRIC_INNER_PRODUCT		)

        SCREAMING_SNAKE_CASE :      Any       =   os.path.join(self.tmpdirname				,				"""hf_bert_base.hnswSQ8_correct_phi_128.c_index"""		)
        dataset.save_faiss_index("""embeddings"""				,				index_file_name + """.index.dpr"""		)
        pickle.dump(dataset["""id"""]				,				open(index_file_name + """.index_meta.dpr"""				,				"""wb"""		)		)

        SCREAMING_SNAKE_CASE :      Optional[Any]       =   os.path.join(self.tmpdirname				,				"""psgs_w100.tsv.pkl"""		)
        SCREAMING_SNAKE_CASE :      Optional[Any]       =   {sample["""id"""]: [sample["""text"""], sample["""title"""]] for sample in dataset}
        pickle.dump(UpperCAmelCase__				,				open(UpperCAmelCase__				,				"""wb"""		)		)

        SCREAMING_SNAKE_CASE :      str       =   RagConfig(
            retrieval_vector_size=self.retrieval_vector_size				,				question_encoder=DPRConfig().to_dict()				,				generator=BartConfig().to_dict()				,				index_name="""legacy"""				,				index_path=self.tmpdirname				,				)
        SCREAMING_SNAKE_CASE :      Any       =   RagRetriever(
            UpperCAmelCase__				,				question_encoder_tokenizer=self.get_dpr_tokenizer()				,				generator_tokenizer=self.get_bart_tokenizer()		)
        return retriever


  def  _lowercase      (			self    : Dict		)					->Optional[int]:
        """simple docstring"""







        SCREAMING_SNAKE_CASE :      Any       =   1
        SCREAMING_SNAKE_CASE :      str       =   self.get_dummy_canonical_hf_index_retriever()
        SCREAMING_SNAKE_CASE :      Union[str, Any]       =   np.array(
            [np.ones(self.retrieval_vector_size		), -np.ones(self.retrieval_vector_size		)]				,				dtype=np.floataa		)
        SCREAMING_SNAKE_CASE     ,	SCREAMING_SNAKE_CASE     ,	SCREAMING_SNAKE_CASE :      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  _lowercase      (			self    : List[str]		)					->int:
        """simple docstring"""







        SCREAMING_SNAKE_CASE :      Optional[Any]       =   self.get_dummy_canonical_hf_index_retriever()
        with tempfile.TemporaryDirectory() as tmp_dirname:
              with patch("""transformers.models.rag.retrieval_rag.load_dataset"""		) as mock_load_dataset:
                    SCREAMING_SNAKE_CASE :      Optional[int]       =   self.get_dummy_dataset()
                    retriever.save_pretrained(UpperCAmelCase__		)
                    SCREAMING_SNAKE_CASE :      List[Any]       =   RagRetriever.from_pretrained(UpperCAmelCase__		)
                    self.assertIsInstance(UpperCAmelCase__				,				UpperCAmelCase__		)
                    SCREAMING_SNAKE_CASE :      Union[str, Any]       =   np.array(
                        [np.ones(self.retrieval_vector_size		), -np.ones(self.retrieval_vector_size		)]				,				dtype=np.floataa		)
                    SCREAMING_SNAKE_CASE :      Dict       =   retriever.retrieve(UpperCAmelCase__				,				n_docs=1		)
                    self.assertTrue(out is not None		)


  def  _lowercase      (			self    : List[Any]		)					->List[str]:
        """simple docstring"""







        SCREAMING_SNAKE_CASE :      Tuple       =   1
        SCREAMING_SNAKE_CASE :      str       =   self.get_dummy_custom_hf_index_retriever(from_disk=UpperCAmelCase__		)
        SCREAMING_SNAKE_CASE :      str       =   np.array(
            [np.ones(self.retrieval_vector_size		), -np.ones(self.retrieval_vector_size		)]				,				dtype=np.floataa		)
        SCREAMING_SNAKE_CASE     ,	SCREAMING_SNAKE_CASE     ,	SCREAMING_SNAKE_CASE :      List[str]       =   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  _lowercase      (			self    : Optional[int]		)					->Optional[Any]:
        """simple docstring"""







        SCREAMING_SNAKE_CASE :      List[str]       =   self.get_dummy_custom_hf_index_retriever(from_disk=UpperCAmelCase__		)
        with tempfile.TemporaryDirectory() as tmp_dirname:
              retriever.save_pretrained(UpperCAmelCase__		)
              SCREAMING_SNAKE_CASE :      Optional[Any]       =   RagRetriever.from_pretrained(UpperCAmelCase__		)
              self.assertIsInstance(UpperCAmelCase__				,				UpperCAmelCase__		)
              SCREAMING_SNAKE_CASE :      Union[str, Any]       =   np.array(
                  [np.ones(self.retrieval_vector_size		), -np.ones(self.retrieval_vector_size		)]				,				dtype=np.floataa		)
              SCREAMING_SNAKE_CASE :      int       =   retriever.retrieve(UpperCAmelCase__				,				n_docs=1		)
              self.assertTrue(out is not None		)


  def  _lowercase      (			self    : Dict		)					->Tuple:
        """simple docstring"""







        SCREAMING_SNAKE_CASE :      Any       =   1
        SCREAMING_SNAKE_CASE :      List[str]       =   self.get_dummy_custom_hf_index_retriever(from_disk=UpperCAmelCase__		)
        SCREAMING_SNAKE_CASE :      Optional[Any]       =   np.array(
            [np.ones(self.retrieval_vector_size		), -np.ones(self.retrieval_vector_size		)]				,				dtype=np.floataa		)
        SCREAMING_SNAKE_CASE     ,	SCREAMING_SNAKE_CASE     ,	SCREAMING_SNAKE_CASE :      List[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  _lowercase      (			self    : Union[str, Any]		)					->Any:
        """simple docstring"""







        SCREAMING_SNAKE_CASE :      Dict       =   self.get_dummy_custom_hf_index_retriever(from_disk=UpperCAmelCase__		)
        with tempfile.TemporaryDirectory() as tmp_dirname:
              retriever.save_pretrained(UpperCAmelCase__		)
              SCREAMING_SNAKE_CASE :      str       =   RagRetriever.from_pretrained(UpperCAmelCase__		)
              self.assertIsInstance(UpperCAmelCase__				,				UpperCAmelCase__		)
              SCREAMING_SNAKE_CASE :      Optional[int]       =   np.array(
                  [np.ones(self.retrieval_vector_size		), -np.ones(self.retrieval_vector_size		)]				,				dtype=np.floataa		)
              SCREAMING_SNAKE_CASE :      Optional[Any]       =   retriever.retrieve(UpperCAmelCase__				,				n_docs=1		)
              self.assertTrue(out is not None		)


  def  _lowercase      (			self    : Tuple		)					->str:
        """simple docstring"""







        SCREAMING_SNAKE_CASE :      Tuple       =   1
        SCREAMING_SNAKE_CASE :      Union[str, Any]       =   self.get_dummy_legacy_index_retriever()
        SCREAMING_SNAKE_CASE :      int       =   np.array(
            [np.ones(self.retrieval_vector_size		), -np.ones(self.retrieval_vector_size		)]				,				dtype=np.floataa		)
        SCREAMING_SNAKE_CASE     ,	SCREAMING_SNAKE_CASE     ,	SCREAMING_SNAKE_CASE :      List[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]		)				,				["""text""", """title"""]		)
        self.assertEqual(len(doc_dicts[0]["""text"""]		)				,				UpperCAmelCase__		)
        self.assertEqual(doc_dicts[0]["""text"""][0]				,				"""bar"""		)  # max inner product is reached with second doc
        self.assertEqual(doc_dicts[1]["""text"""][0]				,				"""foo"""		)  # max inner product is reached with first doc
        self.assertListEqual(doc_ids.tolist()				,				[[1], [0]]		)


  def  _lowercase      (			self    : Any		)					->str:
        """simple docstring"""







        SCREAMING_SNAKE_CASE :      Optional[Any]       =   self.get_dummy_legacy_index_retriever()
        with tempfile.TemporaryDirectory() as tmp_dirname:
              retriever.save_pretrained(UpperCAmelCase__		)
              SCREAMING_SNAKE_CASE :      Union[str, Any]       =   RagRetriever.from_pretrained(UpperCAmelCase__		)
              self.assertIsInstance(UpperCAmelCase__				,				UpperCAmelCase__		)
              SCREAMING_SNAKE_CASE :      str       =   np.array(
                  [np.ones(self.retrieval_vector_size		), -np.ones(self.retrieval_vector_size		)]				,				dtype=np.floataa		)
              SCREAMING_SNAKE_CASE :      Dict       =   retriever.retrieve(UpperCAmelCase__				,				n_docs=1		)
              self.assertTrue(out is not None		)


  @require_torch
  @require_tokenizers
  @require_sentencepiece
  def  _lowercase      (			self    : Tuple		)					->Tuple:
        """simple docstring"""







        import torch

        SCREAMING_SNAKE_CASE :      int       =   1
        SCREAMING_SNAKE_CASE :      Optional[Any]       =   self.get_dummy_canonical_hf_index_retriever()
        SCREAMING_SNAKE_CASE :      str       =   [[5, 7], [1_0, 1_1]]
        SCREAMING_SNAKE_CASE :      Tuple       =   np.array(
            [np.ones(self.retrieval_vector_size		), -np.ones(self.retrieval_vector_size		)]				,				dtype=np.floataa		)
        SCREAMING_SNAKE_CASE :      int       =   retriever(UpperCAmelCase__				,				UpperCAmelCase__				,				prefix=retriever.config.generator.prefix				,				n_docs=UpperCAmelCase__		)
        SCREAMING_SNAKE_CASE     ,	SCREAMING_SNAKE_CASE     ,	SCREAMING_SNAKE_CASE :      List[Any]       =   (
            out["""context_input_ids"""],
            out["""context_attention_mask"""],
            out["""retrieved_doc_embeds"""],
        )
        self.assertEqual(retrieved_doc_embeds.shape				,				(2, n_docs, self.retrieval_vector_size)		)
        self.assertIsInstance(UpperCAmelCase__				,				UpperCAmelCase__		)
        self.assertIsInstance(UpperCAmelCase__				,				UpperCAmelCase__		)
        self.assertIsInstance(UpperCAmelCase__				,				np.ndarray		)

        SCREAMING_SNAKE_CASE :      Union[str, Any]       =   retriever(
            UpperCAmelCase__				,				UpperCAmelCase__				,				prefix=retriever.config.generator.prefix				,				n_docs=UpperCAmelCase__				,				return_tensors="""pt"""				,				)
        SCREAMING_SNAKE_CASE     ,	SCREAMING_SNAKE_CASE     ,	SCREAMING_SNAKE_CASE     ,	SCREAMING_SNAKE_CASE :      Union[str, Any]       =   (  # noqa: F841
            out["""context_input_ids"""],
            out["""context_attention_mask"""],
            out["""retrieved_doc_embeds"""],
            out["""doc_ids"""],
        )
        self.assertEqual(retrieved_doc_embeds.shape				,				(2, n_docs, self.retrieval_vector_size)		)
        self.assertIsInstance(UpperCAmelCase__				,				torch.Tensor		)
        self.assertIsInstance(UpperCAmelCase__				,				torch.Tensor		)
        self.assertIsInstance(UpperCAmelCase__				,				torch.Tensor		)




  @require_torch
  @require_tokenizers
  @require_sentencepiece
  def  _lowercase      (			self    : Union[str, Any]		)					->Optional[Any]:
        """simple docstring"""







        SCREAMING_SNAKE_CASE :      Dict       =   self.get_dpr_ctx_encoder_tokenizer()
        SCREAMING_SNAKE_CASE :      Optional[int]       =   1
        SCREAMING_SNAKE_CASE :      List[Any]       =   self.get_dummy_custom_hf_index_retriever(from_disk=UpperCAmelCase__		)
        retriever.set_ctx_encoder_tokenizer(UpperCAmelCase__		)

        SCREAMING_SNAKE_CASE :      List[str]       =   [[5, 7], [1_0, 1_1]]
        SCREAMING_SNAKE_CASE :      Optional[int]       =   np.array(
            [np.ones(self.retrieval_vector_size		), -np.ones(self.retrieval_vector_size		)]				,				dtype=np.floataa		)
        SCREAMING_SNAKE_CASE :      Union[str, Any]       =   retriever(UpperCAmelCase__				,				UpperCAmelCase__				,				prefix=retriever.config.generator.prefix				,				n_docs=UpperCAmelCase__		)

        self.assertEqual(
            len(UpperCAmelCase__		)				,				6		)  # check whether the retriever output consist of 6 attributes including tokenized docs
        self.assertEqual(
            all(k in out for k in ("""tokenized_doc_ids""", """tokenized_doc_attention_mask""")		)				,				UpperCAmelCase__		)  # check for doc token related keys in dictionary.




