


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 : Optional[Any] 			=				logging.getLogger(__name__)



class    UpperCAmelCase_	(							_a):
       '''simple docstring'''






       def __init__(     self ,					__SCREAMING_SNAKE_CASE ,					__SCREAMING_SNAKE_CASE ,					__SCREAMING_SNAKE_CASE ,					__SCREAMING_SNAKE_CASE=None						):




              """simple docstring"""




              super().__init__(
                  __SCREAMING_SNAKE_CASE ,					question_encoder_tokenizer=__SCREAMING_SNAKE_CASE ,					generator_tokenizer=__SCREAMING_SNAKE_CASE ,					index=__SCREAMING_SNAKE_CASE ,					init_retrieval=__SCREAMING_SNAKE_CASE ,					)
              UpperCamelCase    :	Optional[Any]   	=		None






       def  _lowercase   (     self ,					__SCREAMING_SNAKE_CASE						):




              """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    :	Union[str, Any]   	=		self._infer_socket_ifname()
                     # avoid clash with the NCCL port
                     UpperCamelCase    :	List[str]   	=		str(distributed_port + 1						)
                     UpperCamelCase    :	int   	=		dist.new_group(ranks=__SCREAMING_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						):




              """simple docstring"""




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






       def  _lowercase   (     self ,					__SCREAMING_SNAKE_CASE ,					__SCREAMING_SNAKE_CASE ,					__SCREAMING_SNAKE_CASE=torch.floataa						):




              """simple docstring"""




              UpperCamelCase    :	List[Any]   	=		torch.empty(__SCREAMING_SNAKE_CASE ,					dtype=__SCREAMING_SNAKE_CASE						)
              dist.scatter(__SCREAMING_SNAKE_CASE ,					src=0 ,					scatter_list=__SCREAMING_SNAKE_CASE ,					group=self.process_group						)
              return target_tensor






       def  _lowercase   (     self						):




              """simple docstring"""




              UpperCamelCase    :	int   	=		psutil.net_if_addrs()
              # a hacky way to deal with varying network interface names
              UpperCamelCase    :	Tuple   	=		next((addr for addr in addrs if addr.startswith('''e'''						)) ,					__SCREAMING_SNAKE_CASE						)
              return ifname






       def  _lowercase   (     self ,					__SCREAMING_SNAKE_CASE ,					__SCREAMING_SNAKE_CASE						):




              """simple docstring"""




              if not dist.is_initialized():
                     UpperCamelCase	, UpperCamelCase    :	List[str]   	=		self._main_retrieve(__SCREAMING_SNAKE_CASE ,					__SCREAMING_SNAKE_CASE						)
                     return retrieved_doc_embeds, doc_ids, self.index.get_doc_dicts(__SCREAMING_SNAKE_CASE						)

              # distributed training
              UpperCamelCase    :	str   	=		dist.get_world_size(group=self.process_group						)

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

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

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





def      a	(							SCREAMING_SNAKE_CASE_      :		int							,			SCREAMING_SNAKE_CASE_      :		int       ):






       """simple docstring"""





       if a < 0 or b < 0:
              raise ValueError('''the value of both inputs must be positive'''       )
       UpperCamelCase    :	int   	=		str(bin(SCREAMING_SNAKE_CASE_       )       )[2:]  # remove the leading "0b"
       UpperCamelCase    :	List[str]   	=		str(bin(SCREAMING_SNAKE_CASE_       )       )[2:]
       UpperCamelCase    :	Tuple   	=		max(len(SCREAMING_SNAKE_CASE_       )							,			len(SCREAMING_SNAKE_CASE_       )       )
       return "0b" + "".join(
           str(int('''1''' in (char_a, char_b)       )       )
           for char_a, char_b in zip(a_binary.zfill(SCREAMING_SNAKE_CASE_       )							,			b_binary.zfill(SCREAMING_SNAKE_CASE_       )       )       )


if __name__ == "__main__":
  import doctest

  doctest.testmod()

