





'''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        =						logging.getLogger(__name__)







class __a (   snake_case__			):





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

      '''simple docstring'''

      super().__init__(
          _A				,question_encoder_tokenizer=_A				,generator_tokenizer=_A				,index=_A				,init_retrieval=_A				,)
      __SCREAMING_SNAKE_CASE      							=       None





   def      UpperCAmelCase__     (					self				:			int				,lowerCamelCase				:			List[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
         __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=_A				,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      UpperCAmelCase__     (					self				:			int    ):

      '''simple docstring'''

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





   def      UpperCAmelCase__     (					self				:			Optional[Any]				,lowerCamelCase				:			List[Any]				,lowerCamelCase				:			Tuple				,lowerCamelCase				:			Optional[int]=torch.floataa    ):

      '''simple docstring'''

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





   def      UpperCAmelCase__     (					self				:			Union[str, Any]    ):

      '''simple docstring'''

      __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"""    ))				,_A    )
      return ifname





   def      UpperCAmelCase__     (					self				:			str				,lowerCamelCase				:			List[Any]				,lowerCamelCase				:			Union[str, Any]    ):

      '''simple docstring'''

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

      # 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(_A    )]
      dist.gather(torch.tensor(_A    )				,dst=0				,gather_list=_A				,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(_A    ) == world_size
         __SCREAMING_SNAKE_CASE      							=       self._main_retrieve(torch.cat(_A    ).numpy()				,_A    )
         __SCREAMING_SNAKE_CASE      							=       torch.tensor(_A    ), torch.tensor(_A    )
         __SCREAMING_SNAKE_CASE      							=       self._chunk_tensor(_A				,_A    )
         __SCREAMING_SNAKE_CASE      							=       self._chunk_tensor(_A				,_A    )
      __SCREAMING_SNAKE_CASE      							=       self._scattered(_A				,[n_queries, n_docs]				,target_type=torch.intaa    )
      __SCREAMING_SNAKE_CASE      							=       self._scattered(_A				,[n_queries, n_docs, question_hidden_states.shape[1]]    )

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












'''simple docstring'''

from .dependency_versions_table import deps
from .utils.versions import require_version, require_version_core


# define which module versions we always want to check at run time
# (usually the ones defined in `install_requires` in setup.py)
#
# order specific notes:
# - tqdm must be checked before tokenizers

a        =						[
    "python",
    "tqdm",
    "regex",
    "requests",
    "packaging",
    "filelock",
    "numpy",
    "tokenizers",
    "huggingface-hub",
    "safetensors",
    "accelerate",
    "pyyaml",
]

for pkg in pkgs_to_check_at_runtime:
    if pkg in deps:
        if pkg == "tokenizers":
            # must be loaded here, or else tqdm check may fail
            from .utils import is_tokenizers_available

            if not is_tokenizers_available():
                continue  # not required, check version only if installed
        elif pkg == "accelerate":
            # must be loaded here, or else tqdm check may fail
            from .utils import is_accelerate_available

            # Maybe switch to is_torch_available in the future here so that Accelerate is hard dep of
            # Transformers with PyTorch
            if not is_accelerate_available():
                continue  # not required, check version only if installed

        require_version_core(deps[pkg])
    else:
        raise ValueError(F'''can\'t find {pkg} in {deps.keys()}, check dependency_versions_table.py''')

def 						__magic_name__       (							__UpperCAmelCase							,	__UpperCAmelCase=None      )						->					Optional[Any]:

   '''simple docstring'''







   require_version(deps[pkg]							,	__UpperCAmelCase      )





