

"""simple docstring"""




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


__UpperCAmelCase =				logging.getLogger(__name__)







class      __lowercase			(    __lowerCamelCase					):

  def __init__(  self		:			Any					,A		:			List[str]					,A		:			List[str]					,A		:			int					,A		:			Optional[Any]=None    ):






   '''simple docstring'''
   super().__init__(
       UpperCamelCase_					,question_encoder_tokenizer=UpperCamelCase_					,generator_tokenizer=UpperCamelCase_					,index=UpperCamelCase_					,init_retrieval=UpperCamelCase_					,)
   UpperCAmelCase__					:  Optional[int]					   =					None

  def    __lowercase       (  self		:			str					,A		:			int    ):






   '''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
    UpperCAmelCase__					:  int					   =					self._infer_socket_ifname()
    # avoid clash with the NCCL port
    UpperCAmelCase__					:  Union[str, Any]					   =					str(distributed_port + 1    )
    UpperCAmelCase__					:  Optional[int]					   =					dist.new_group(ranks=UpperCamelCase_					,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    __lowercase       (  self		:			Dict    ):






   '''simple docstring'''
   return dist.get_rank(group=self.process_group    ) == 0

  def    __lowercase       (  self		:			Union[str, Any]					,A		:			int					,A		:			Tuple					,A		:			Optional[Any]=torch.floataa    ):






   '''simple docstring'''
   UpperCAmelCase__					:  Union[str, Any]					   =					torch.empty(UpperCamelCase_					,dtype=UpperCamelCase_    )
   dist.scatter(UpperCamelCase_					,src=0					,scatter_list=UpperCamelCase_					,group=self.process_group    )
   return target_tensor

  def    __lowercase       (  self		:			str    ):






   '''simple docstring'''
   UpperCAmelCase__					:  Union[str, Any]					   =					psutil.net_if_addrs()
   # a hacky way to deal with varying network interface names
   UpperCAmelCase__					:  List[str]					   =					next((addr for addr in addrs if addr.startswith("""e"""    ))					,UpperCamelCase_    )
   return ifname




  def    __lowercase       (  self		:			str					,A		:			np.ndarray					,A		:			int    ):






   '''simple docstring'''





   if not dist.is_initialized():
    UpperCAmelCase__					:  Optional[Any]					   =					self._main_retrieve(UpperCamelCase_					,UpperCamelCase_    )
    return retrieved_doc_embeds, doc_ids, self.index.get_doc_dicts(UpperCamelCase_    )

   # distributed training
   UpperCAmelCase__					:  int					   =					dist.get_world_size(group=self.process_group    )

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

   # scatter logic
   UpperCAmelCase__					:  Optional[Any]					   =					question_hidden_states.shape[0]
   UpperCAmelCase__					:  Optional[Any]					   =					[]
   UpperCAmelCase__					:  int					   =					[]
   if self._is_main():
    assert len(UpperCamelCase_    ) == world_size
    UpperCAmelCase__					:  Any					   =					self._main_retrieve(torch.cat(UpperCamelCase_    ).numpy()					,UpperCamelCase_    )
    UpperCAmelCase__					:  int					   =					torch.tensor(UpperCamelCase_    ), torch.tensor(UpperCamelCase_    )
    UpperCAmelCase__					:  str					   =					self._chunk_tensor(UpperCamelCase_					,UpperCamelCase_    )
    UpperCAmelCase__					:  int					   =					self._chunk_tensor(UpperCamelCase_					,UpperCamelCase_    )
   UpperCAmelCase__					:  List[Any]					   =					self._scattered(UpperCamelCase_					,[n_queries, n_docs]					,target_type=torch.intaa    )
   UpperCAmelCase__					:  List[Any]					   =					self._scattered(UpperCamelCase_					,[n_queries, n_docs, question_hidden_states.shape[1]]    )

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








"""simple docstring"""




# tests directory-specific settings - this file is run automatically
# by pytest before any tests are run

import sys
import warnings
from os.path import abspath, dirname, join


# allow having multiple repository checkouts and not needing to remember to rerun
# 'pip install -e .[dev]' when switching between checkouts and running tests.
__UpperCAmelCase =				abspath(join(dirname(dirname(__file__)), 'src'))
sys.path.insert(1, git_repo_path)

# silence FutureWarning warnings in tests since often we can't act on them until
# they become normal warnings - i.e. the tests still need to test the current functionality
warnings.simplefilter(action='ignore', category=FutureWarning)
def   lowerCAmelCase						(					__UpperCamelCase      ):
 '''simple docstring'''





 from diffusers.utils.testing_utils import pytest_addoption_shared

 pytest_addoption_shared(__UpperCamelCase      )




def   lowerCAmelCase						(					__UpperCamelCase      ):
 '''simple docstring'''





 from diffusers.utils.testing_utils import pytest_terminal_summary_main

 UpperCAmelCase__					:  Optional[Any]					   =					terminalreporter.config.getoption("""--make-reports"""      )
 if make_reports:
  pytest_terminal_summary_main(__UpperCamelCase							,    id=__UpperCamelCase      )





