
import logging
import os
from typing import List, Tuple

import numpy as np
import psutil
import torch
import torch.distributed as dist

from transformers import RagRetriever


A_		:     str        =     logging.getLogger(__name__)



class 		_lowercase		(       UpperCAmelCase__	):

   def __init__(      self			:       Any							,   __lowerCAmelCase			:       str							,   __lowerCAmelCase			:       List[Any]							,   __lowerCAmelCase			:       str							,   __lowerCAmelCase			:       Tuple=None							)       ->				Optional[int]:


    """simple docstring"""






    super().__init__(
        __lowerCAmelCase							,   question_encoder_tokenizer=__lowerCAmelCase							,   generator_tokenizer=__lowerCAmelCase							,   index=__lowerCAmelCase							,   init_retrieval=__lowerCAmelCase							,   )
    a						=		None

   def 		A							(      self			:       Optional[int]							,   __lowerCAmelCase			:       int							)       ->				Union[str, Any]:


    """simple docstring"""






    logger.info("initializing retrieval"							)

    # initializing a separate process group for retrieval as the default
    # nccl backend doesn't support gather/scatter operations while gloo
    # is too slow to replace nccl for the core gpu communication
    if dist.is_initialized():
     logger.info("dist initialized"							)
     # needs to be set manually
     a						=		self._infer_socket_ifname()
     # avoid clash with the NCCL port
     a						=		str(distributed_port + 1							)
     a						=		dist.new_group(ranks=__lowerCAmelCase							,   backend="gloo"							)

    # initialize retriever only on the main worker
    if not dist.is_initialized() or self._is_main():
     logger.info("dist not initialized / main"							)
     self.index.init_index()

    # all processes wait untill the retriever is initialized by the main process
    if dist.is_initialized():
     torch.distributed.barrier(group=self.process_group							)

   def 		A							(      self			:       List[Any]							)       ->				Union[str, Any]:


    """simple docstring"""






    return dist.get_rank(group=self.process_group							) == 0

   def 		A							(      self			:       List[str]							,   __lowerCAmelCase			:       List[Any]							,   __lowerCAmelCase			:       Any							,   __lowerCAmelCase			:       Tuple=torch.floataa							)       ->				Dict:


    """simple docstring"""






    a						=		torch.empty(__lowerCAmelCase							,   dtype=__lowerCAmelCase							)
    dist.scatter(__lowerCAmelCase							,   src=0							,   scatter_list=__lowerCAmelCase							,   group=self.process_group							)
    return target_tensor

   def 		A							(      self			:       Dict							)       ->				List[str]:


    """simple docstring"""






    a						=		psutil.net_if_addrs()
    # a hacky way to deal with varying network interface names
    a						=		next((addr for addr in addrs if addr.startswith("e"							))							,   __lowerCAmelCase							)
    return ifname




   def 		A							(      self			:       str							,   __lowerCAmelCase			:       np.ndarray							,   __lowerCAmelCase			:       int							)       ->				Tuple[np.ndarray, List[dict]]:


    """simple docstring"""






    if not dist.is_initialized():
     a		,  a						=		self._main_retrieve(__lowerCAmelCase							,   __lowerCAmelCase							)
     return retrieved_doc_embeds, doc_ids, self.index.get_doc_dicts(__lowerCAmelCase							)

    # distributed training
    a						=		dist.get_world_size(group=self.process_group							)

    # gather logic
    a						=		None
    if self._is_main():
     a						=		[torch.empty(question_hidden_states.shape							,   dtype=torch.floataa							) for _ in range(__lowerCAmelCase							)]
    dist.gather(torch.tensor(__lowerCAmelCase							)							,   dst=0							,   gather_list=__lowerCAmelCase							,   group=self.process_group							)

    # scatter logic
    a						=		question_hidden_states.shape[0]
    a						=		[]
    a						=		[]
    if self._is_main():
     assert len(__lowerCAmelCase							) == world_size
     a		,  a						=		self._main_retrieve(torch.cat(__lowerCAmelCase							).numpy()							,   __lowerCAmelCase							)
     a		,  a						=		torch.tensor(__lowerCAmelCase							), torch.tensor(__lowerCAmelCase							)
     a						=		self._chunk_tensor(__lowerCAmelCase							,   __lowerCAmelCase							)
     a						=		self._chunk_tensor(__lowerCAmelCase							,   __lowerCAmelCase							)
    a						=		self._scattered(__lowerCAmelCase							,   [n_queries, n_docs]							,   target_type=torch.intaa							)
    a						=		self._scattered(__lowerCAmelCase							,   [n_queries, n_docs, question_hidden_states.shape[1]]							)

    return retrieved_doc_embeds.numpy(), doc_ids.numpy(), self.index.get_doc_dicts(__lowerCAmelCase							)






from ...utils import (
    OptionalDependencyNotAvailable,
    is_torch_available,
    is_transformers_available,
    is_transformers_version,
)


try:
     if not (is_transformers_available() and is_torch_available() and is_transformers_version('''>=''', '''4.25.0''')):
          raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
     from ...utils.dummy_torch_and_transformers_objects import UnCLIPImageVariationPipeline, UnCLIPPipeline
else:
     from .pipeline_unclip import UnCLIPPipeline
     from .pipeline_unclip_image_variation import UnCLIPImageVariationPipeline
     from .text_proj import UnCLIPTextProjModel

