




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


lowerCamelCase__			:       int         =							logging.getLogger(__name__)





class 	lowerCamelCase_							(				_SCREAMING_SNAKE_CASE ):


 '''simple docstring'''







 def __init__(							self  :		Optional[Any]    ,    _lowerCAmelCase  :		str    ,    _lowerCAmelCase  :		Optional[int]    ,    _lowerCAmelCase  :		int    ,    _lowerCAmelCase  :		Optional[int]=None						):
       super().__init__(
           _lowerCAmelCase    ,    question_encoder_tokenizer=_lowerCAmelCase    ,    generator_tokenizer=_lowerCAmelCase    ,    index=_lowerCAmelCase    ,    init_retrieval=_lowerCAmelCase    ,    )
       SCREAMING_SNAKE_CASE_     						=      None



 def 				lowerCAmelCase_  (							self  :		Any    ,    _lowerCAmelCase  :		int						):
       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
             SCREAMING_SNAKE_CASE_     						=      self._infer_socket_ifname()
             # avoid clash with the NCCL port
             SCREAMING_SNAKE_CASE_     						=      str(distributed_port + 1						)
             SCREAMING_SNAKE_CASE_     						=      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 				lowerCAmelCase_  (							self  :		Any						):
       return dist.get_rank(group=self.process_group						) == 0



 def 				lowerCAmelCase_  (							self  :		Optional[Any]    ,    _lowerCAmelCase  :		Dict    ,    _lowerCAmelCase  :		Any    ,    _lowerCAmelCase  :		Any=torch.floataa						):
       SCREAMING_SNAKE_CASE_     						=      torch.empty(_lowerCAmelCase    ,    dtype=_lowerCAmelCase						)
       dist.scatter(_lowerCAmelCase    ,    src=0    ,    scatter_list=_lowerCAmelCase    ,    group=self.process_group						)
       return target_tensor



 def 				lowerCAmelCase_  (							self  :		List[str]						):
       SCREAMING_SNAKE_CASE_     						=      psutil.net_if_addrs()
       # a hacky way to deal with varying network interface names
       SCREAMING_SNAKE_CASE_     						=      next((addr for addr in addrs if addr.startswith('e'						))    ,    _lowerCAmelCase						)
       return ifname




 def 				lowerCAmelCase_  (							self  :		Optional[int]    ,    _lowerCAmelCase  :		np.ndarray    ,    _lowerCAmelCase  :		int						):

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

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

       # gather logic
       SCREAMING_SNAKE_CASE_     						=      None
       if self._is_main():
             SCREAMING_SNAKE_CASE_     						=      [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
       SCREAMING_SNAKE_CASE_     						=      question_hidden_states.shape[0]
       SCREAMING_SNAKE_CASE_     						=      []
       SCREAMING_SNAKE_CASE_     						=      []
       if self._is_main():
             assert len(_lowerCAmelCase						) == world_size
             SCREAMING_SNAKE_CASE_						,			SCREAMING_SNAKE_CASE_     						=      self._main_retrieve(torch.cat(_lowerCAmelCase						).numpy()    ,    _lowerCAmelCase						)
             SCREAMING_SNAKE_CASE_						,			SCREAMING_SNAKE_CASE_     						=      torch.tensor(_lowerCAmelCase						), torch.tensor(_lowerCAmelCase						)
             SCREAMING_SNAKE_CASE_     						=      self._chunk_tensor(_lowerCAmelCase    ,    _lowerCAmelCase						)
             SCREAMING_SNAKE_CASE_     						=      self._chunk_tensor(_lowerCAmelCase    ,    _lowerCAmelCase						)
       SCREAMING_SNAKE_CASE_     						=      self._scattered(_lowerCAmelCase    ,    [n_queries, n_docs]    ,    target_type=torch.intaa						)
       SCREAMING_SNAKE_CASE_     						=      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 abc import ABC, abstractmethod
from argparse import ArgumentParser





class 	lowerCamelCase_							(				_SCREAMING_SNAKE_CASE ):


 '''simple docstring'''







 @staticmethod
 @abstractmethod
 def 				lowerCAmelCase_  (							_lowerCAmelCase  :		ArgumentParser						):
       raise NotImplementedError()




 @abstractmethod
 def 				lowerCAmelCase_  (							self  :		Dict						):
       raise NotImplementedError()