




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 (
    center_crop,
    convert_to_rgb,
    get_resize_output_image_size,
    normalize,
    rescale,
    resize,
    to_channel_dimension_format,
)
from ...image_utils import (
    OPENAI_CLIP_MEAN,
    OPENAI_CLIP_STD,
    ChannelDimension,
    ImageInput,
    PILImageResampling,
    make_list_of_images,
    to_numpy_array,
    valid_images,
)
from ...utils import TensorType, is_vision_available, logging


UpperCamelCase__       				=	logging.get_logger(__name__)


if is_vision_available():
  import PIL
class 							__lowercase   (			a__    ):
     _lowerCAmelCase		       =						["pixel_values"]



     def __init__(  self   :							List[str]      ,		lowercase__   :							bool = True      ,		lowercase__   :							Dict[str, int] = None      ,		lowercase__   :							PILImageResampling = PILImageResampling.BICUBIC      ,		lowercase__   :							bool = True      ,		lowercase__   :							Dict[str, int] = None      ,		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__   :							bool = True      ,		**lowercase__   :							str      ,		):
       super().__init__(**lowercase__       )
       a_		    =  size if size is not None else {'''shortest_edge''': 2_2_4}
       a_		    =  get_size_dict(lowercase__      ,		default_to_square=lowercase__       )
       a_		    =  crop_size if crop_size is not None else {'''height''': 2_2_4, '''width''': 2_2_4}
       a_		    =  get_size_dict(lowercase__       )

       a_		    =  do_resize
       a_		    =  size
       a_		    =  resample
       a_		    =  do_center_crop
       a_		    =  crop_size
       a_		    =  do_rescale
       a_		    =  rescale_factor
       a_		    =  do_normalize
       a_		    =  image_mean if image_mean is not None else OPENAI_CLIP_MEAN
       a_		    =  image_std if image_std is not None else OPENAI_CLIP_STD
       a_		    =  do_convert_rgb



     def 				__magic_name__		(  self   :							List[str]      ,		lowercase__   :							np.ndarray      ,		lowercase__   :							Dict[str, int]      ,		lowercase__   :							PILImageResampling = PILImageResampling.BICUBIC      ,		lowercase__   :							Optional[Union[str, ChannelDimension]] = None      ,		**lowercase__   :							Any      ,		):
       a_		    =  get_size_dict(lowercase__      ,		default_to_square=lowercase__       )
       a_		    =  get_resize_output_image_size(
           lowercase__      ,		size=(size['''height'''], size['''width'''])      ,		default_to_square=lowercase__       )
       return resize(lowercase__      ,		size=lowercase__      ,		resample=lowercase__      ,		data_format=lowercase__      ,		**lowercase__       )



     def 				__magic_name__		(  self   :							List[Any]      ,		lowercase__   :							np.ndarray      ,		lowercase__   :							Dict[str, int]      ,		lowercase__   :							Optional[Union[str, ChannelDimension]] = None      ,		**lowercase__   :							str      ,		):
       a_		    =  get_size_dict(lowercase__       )
       return center_crop(lowercase__      ,		size=(size['''height'''], size['''width'''])      ,		data_format=lowercase__      ,		**lowercase__       )



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



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



     def 				__magic_name__		(  self   :							str      ,		lowercase__   :							ImageInput      ,		lowercase__   :							bool = None      ,		lowercase__   :							Dict[str, int] = None      ,		lowercase__   :							PILImageResampling = None      ,		lowercase__   :							bool = None      ,		lowercase__   :							int = None      ,		lowercase__   :							bool = None      ,		lowercase__   :							float = None      ,		lowercase__   :							bool = None      ,		lowercase__   :							Optional[Union[float, List[float]]] = None      ,		lowercase__   :							Optional[Union[float, List[float]]] = None      ,		lowercase__   :							bool = None      ,		lowercase__   :							Optional[Union[str, TensorType]] = None      ,		lowercase__   :							Optional[ChannelDimension] = ChannelDimension.FIRST      ,		**lowercase__   :							int      ,		):
       a_		    =  do_resize if do_resize is not None else self.do_resize
       a_		    =  size if size is not None else self.size
       a_		    =  get_size_dict(lowercase__      ,		default_to_square=lowercase__       )
       a_		    =  resample if resample is not None else self.resample
       a_		    =  do_center_crop if do_center_crop is not None else self.do_center_crop
       a_		    =  crop_size if crop_size is not None else self.crop_size
       a_		    =  get_size_dict(lowercase__       )
       a_		    =  do_rescale if do_rescale is not None else self.do_rescale
       a_		    =  rescale_factor if rescale_factor is not None else self.rescale_factor
       a_		    =  do_normalize if do_normalize is not None else self.do_normalize
       a_		    =  image_mean if image_mean is not None else self.image_mean
       a_		    =  image_std if image_std is not None else self.image_std
       a_		    =  do_convert_rgb if do_convert_rgb is not None else self.do_convert_rgb

       a_		    =  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_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.'''       )

       # PIL RGBA images are converted to RGB
       if do_convert_rgb:
         a_		    =  [convert_to_rgb(lowercase__       ) for image in images]

       # All transformations expect numpy arrays.
       a_		    =  [to_numpy_array(lowercase__       ) for image in images]

       if do_resize:
         a_		    =  [self.resize(image=lowercase__      ,		size=lowercase__      ,		resample=lowercase__       ) for image in images]

       if do_center_crop:
         a_		    =  [self.center_crop(image=lowercase__      ,		size=lowercase__       ) for image in images]

       if do_rescale:
         a_		    =  [self.rescale(image=lowercase__      ,		scale=lowercase__       ) for image in images]

       if do_normalize:
         a_		    =  [self.normalize(image=lowercase__      ,		mean=lowercase__      ,		std=lowercase__       ) for image in images]

       a_		    =  [to_channel_dimension_format(lowercase__      ,		lowercase__       ) for image in images]

       a_		    =  {'''pixel_values''': images}
       return BatchFeature(data=lowercase__      ,		tensor_type=lowercase__       )






import logging
import random

import ray

from transformers import RagConfig, RagRetriever, RagTokenizer
from transformers.models.rag.retrieval_rag import CustomHFIndex


UpperCamelCase__       				=	logging.getLogger(__name__)
class 							__lowercase   :



     def __init__(  self   :							List[str]       ):
       a_		    =  False



     def 				__magic_name__		(  self   :							Union[str, Any]      ,		lowercase__   :							List[str]      ,		lowercase__   :							Optional[int]      ,		lowercase__   :							Optional[Any]      ,		lowercase__   :							Union[str, Any]       ):
       if not self.initialized:
         a_		    =  RagRetriever(
             lowercase__      ,		question_encoder_tokenizer=lowercase__      ,		generator_tokenizer=lowercase__      ,		index=lowercase__      ,		init_retrieval=lowercase__      ,		)
         a_		    =  True



     def 				__magic_name__		(  self   :							List[Any]       ):
       self.retriever.index.init_index()



     def 				__magic_name__		(  self   :							List[Any]       ):
       # delete the old self.retriever object before assigning the new index
       del self.retriever
       a_		    =  False



     def 				__magic_name__		(  self   :							Optional[Any]      ,		lowercase__   :							List[str]      ,		lowercase__   :							Dict       ):
       a_    , a_		    =  self.retriever._main_retrieve(lowercase__      ,		lowercase__       )
       a_		    =  self.retriever.index.get_doc_dicts(lowercase__       )
       return doc_ids, retrieved_doc_embeds, doc_dicts



class 							__lowercase   (			a__    ):



     def __init__(  self   :							List[Any]      ,		lowercase__   :							Optional[int]      ,		lowercase__   :							List[str]      ,		lowercase__   :							Tuple      ,		lowercase__   :							Dict      ,		lowercase__   :							Dict=None       ):
       if index is not None and index.is_initialized() and len(lowercase__       ) > 0:
         raise ValueError(
             '''When using Ray for distributed fine-tuning, '''
             '''you\'ll need to provide the paths instead, '''
             '''as the dataset and the index are loaded '''
             '''separately. More info in examples/rag/use_own_knowledge_dataset.py '''       )

       super().__init__(
           lowercase__      ,		question_encoder_tokenizer=lowercase__      ,		generator_tokenizer=lowercase__      ,		index=lowercase__      ,		init_retrieval=lowercase__      ,		)

       a_		    =  retrieval_workers
       a_		    =  question_encoder_tokenizer
       a_		    =  generator_tokenizer
       if len(self.retrieval_workers       ) > 0:
         ray.get(
             [
                 worker.create_rag_retriever.remote(lowercase__      ,		lowercase__      ,		lowercase__      ,		lowercase__       )
                 for worker in self.retrieval_workers
             ]       )



     def 				__magic_name__		(  self   :							int       ):
       logger.info('''initializing retrieval'''       )

       if len(self.retrieval_workers       ) > 0:
         ray.get([worker.init_retrieval.remote() for worker in self.retrieval_workers]       )
       else:
         # Non-distributed training. Load index into this same process.
         self.index.init_index()



     def 				__magic_name__		(  self   :							str      ,		lowercase__   :							List[str]      ,		lowercase__   :							Any       ):
       if len(self.retrieval_workers       ) > 0:
         # Select a random retrieval actor.
         a_		    =  self.retrieval_workers[random.randint(0      ,		len(self.retrieval_workers       ) - 1       )]
         a_    , a_    , a_		    =  ray.get(
             random_worker.retrieve.remote(lowercase__      ,		lowercase__       )       )
       else:
         a_    , a_		    =  self._main_retrieve(lowercase__      ,		lowercase__       )
         a_		    =  self.index.get_doc_dicts(lowercase__       )
       return retrieved_doc_embeds, doc_ids, doc_dicts



     @classmethod
     def 				__magic_name__		(  cls   :							Dict      ,		lowercase__   :							Optional[Any]      ,		lowercase__   :							List[str]=None      ,		**lowercase__   :							int       ):
       return super(lowercase__      ,		cls       ).get_tokenizers(lowercase__      ,		lowercase__      ,		**lowercase__       )



     @classmethod
     def 				__magic_name__		(  cls   :							Tuple      ,		lowercase__   :							Any      ,		lowercase__   :							str      ,		lowercase__   :							List[Any]=None      ,		**lowercase__   :							Union[str, Any]       ):
       a_		    =  kwargs.pop('''config'''      ,		lowercase__       ) or RagConfig.from_pretrained(lowercase__      ,		**lowercase__       )
       a_		    =  RagTokenizer.from_pretrained(lowercase__      ,		config=lowercase__       )
       a_		    =  rag_tokenizer.question_encoder
       a_		    =  rag_tokenizer.generator

       if indexed_dataset is not None:
         a_		    =  '''custom'''
         a_		    =  CustomHFIndex(config.retrieval_vector_size      ,		lowercase__       )
       else:
         a_		    =  cls._build_index(lowercase__       )

       return cls(
           lowercase__      ,		question_encoder_tokenizer=lowercase__      ,		generator_tokenizer=lowercase__      ,		retrieval_workers=lowercase__      ,		index=lowercase__      ,		)



     def 				__magic_name__		(  self   :							Optional[Any]       ):
       logger.info('''re-loading the new dataset with embeddings'''       )
       # access from the training loop

       ray.get([worker.clear_object.remote() for worker in self.retrieval_workers]       )

       # build the index object again
       a_		    =  self._build_index(self.config       )

       ray.get(
           [
               worker.create_rag_retriever.remote(
                   self.config      ,		self.question_encoder_tokenizer      ,		self.generator_tokenizer      ,		lowercase__       )
               for worker in self.retrieval_workers
           ]       )
