





'''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


A__				:   Tuple      =logging.getLogger(__name__)




class  UpperCAmelCase     (			snake_case_   ):




      def __init__(			self			: int			,			__snake_case			: Union[str, Any]			,			__snake_case			: Dict			,			__snake_case			: Any			,			__snake_case			: str=None  )				->      Union[str, Any]:
            super().__init__(
                __snake_case			,			question_encoder_tokenizer=__snake_case			,			generator_tokenizer=__snake_case			,			index=__snake_case			,			init_retrieval=__snake_case			,			)
            _lowerCAmelCase							     =							None




      def 				lowercase__    (			self			: List[Any]			,			__snake_case			: int  )				->      Optional[Any]:
            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
                  _lowerCAmelCase							     =							self._infer_socket_ifname()
                  # avoid clash with the NCCL port
                  _lowerCAmelCase							     =							str(distributed_port + 1  )
                  _lowerCAmelCase							     =							dist.new_group(ranks=__snake_case			,			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			: str  )				->      Any:
            return dist.get_rank(group=self.process_group  ) == 0




      def 				lowercase__    (			self			: str			,			__snake_case			: str			,			__snake_case			: List[Any]			,			__snake_case			: str=torch.floataa  )				->      List[str]:
            _lowerCAmelCase							     =							torch.empty(__snake_case			,			dtype=__snake_case  )
            dist.scatter(__snake_case			,			src=0			,			scatter_list=__snake_case			,			group=self.process_group  )
            return target_tensor




      def 				lowercase__    (			self			: Optional[Any]  )				->      int:
            _lowerCAmelCase							     =							psutil.net_if_addrs()
            # a hacky way to deal with varying network interface names
            _lowerCAmelCase							     =							next((addr for addr in addrs if addr.startswith("""e"""  ))			,			__snake_case  )
            return ifname




      def 				lowercase__    (			self			: Optional[int]			,			__snake_case			: np.ndarray			,			__snake_case			: int  )				->      Tuple[np.ndarray, List[dict]]:

            # single GPU training
            if not dist.is_initialized():
                  _lowerCAmelCase     ,							_lowerCAmelCase							     =							self._main_retrieve(__snake_case			,			__snake_case  )
                  return retrieved_doc_embeds, doc_ids, self.index.get_doc_dicts(__snake_case  )

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

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

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

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











"""simple docstring"""







from ..utils import DummyObject, requires_backends
class        __A       (	metaclass=A_    ):







		'''simple docstring'''





		lowerCAmelCase			:					List[str]   	=				["torch", "torchsde"]







		def __init__(	self						:  Tuple			,*_snake_case						:  Union[str, Any]			,**_snake_case						:  Any						)					->		Union[str, Any]:






				"""simple docstring"""



				requires_backends(self			,['''torch''', '''torchsde''']						)







		@classmethod
		def 					UpperCAmelCase    (	cls						:  List[str]			,*_snake_case						:  int			,**_snake_case						:  Union[str, Any]						)					->		str:






				"""simple docstring"""



				requires_backends(cls			,['''torch''', '''torchsde''']						)







		@classmethod
		def 					UpperCAmelCase    (	cls						:  List[Any]			,*_snake_case						:  List[Any]			,**_snake_case						:  List[str]						)					->		List[Any]:






				"""simple docstring"""



				requires_backends(cls			,['''torch''', '''torchsde''']						)
