



"""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 					a	(  UpperCAmelCase__       ):



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


     '''simple docstring'''




     super().__init__(
         __lowerCAmelCase       ,		question_encoder_tokenizer=__lowerCAmelCase       ,		generator_tokenizer=__lowerCAmelCase       ,		index=__lowerCAmelCase       ,		init_retrieval=__lowerCAmelCase       ,		)
     SCREAMING_SNAKE_CASE_: List[str]     			=None



  def 	lowerCamelCase__       (						self :					Optional[int]       ,		lowerCAmelCase :					int		)     ->					Union[str, 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_: List[str]     			=self._infer_socket_ifname()
        # avoid clash with the NCCL port
        SCREAMING_SNAKE_CASE_: Optional[int]     			=str(distributed_port + 1		)
        SCREAMING_SNAKE_CASE_: Any     			=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 :					List[Any]		)     ->					Union[str, Any]:


     '''simple docstring'''




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



  def 	lowerCamelCase__       (						self :					List[str]       ,		lowerCAmelCase :					List[Any]       ,		lowerCAmelCase :					Any       ,		lowerCAmelCase :					Tuple=torch.floataa		)     ->					Dict:


     '''simple docstring'''




     SCREAMING_SNAKE_CASE_: Dict     			=torch.empty(__lowerCAmelCase       ,		dtype=__lowerCAmelCase		)
     dist.scatter(__lowerCAmelCase       ,		src=0       ,		scatter_list=__lowerCAmelCase       ,		group=self.process_group		)
     return target_tensor



  def 	lowerCamelCase__       (						self :					Dict		)     ->					List[str]:


     '''simple docstring'''




     SCREAMING_SNAKE_CASE_: int     			=psutil.net_if_addrs()
     # a hacky way to deal with varying network interface names
     SCREAMING_SNAKE_CASE_: str     			=next((addr for addr in addrs if addr.startswith("""e"""		))       ,		__lowerCAmelCase		)
     return ifname





  def 	lowerCamelCase__       (						self :					str       ,		lowerCAmelCase :					np.ndarray       ,		lowerCAmelCase :					int		)     ->					Tuple[np.ndarray, List[dict]]:


     '''simple docstring'''




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

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

     # gather logic
     SCREAMING_SNAKE_CASE_: List[str]     			=None
     if self._is_main():
        SCREAMING_SNAKE_CASE_: Dict     			=[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_: int     			=question_hidden_states.shape[0]
     SCREAMING_SNAKE_CASE_: Dict     			=[]
     SCREAMING_SNAKE_CASE_: Tuple     			=[]
     if self._is_main():
        assert len(__lowerCAmelCase		) == world_size
        SCREAMING_SNAKE_CASE_			,		SCREAMING_SNAKE_CASE_: List[str]     			=self._main_retrieve(torch.cat(__lowerCAmelCase		).numpy()       ,		__lowerCAmelCase		)
        SCREAMING_SNAKE_CASE_			,		SCREAMING_SNAKE_CASE_: Optional[int]     			=torch.tensor(__lowerCAmelCase		), torch.tensor(__lowerCAmelCase		)
        SCREAMING_SNAKE_CASE_: Dict     			=self._chunk_tensor(__lowerCAmelCase       ,		__lowerCAmelCase		)
        SCREAMING_SNAKE_CASE_: Tuple     			=self._chunk_tensor(__lowerCAmelCase       ,		__lowerCAmelCase		)
     SCREAMING_SNAKE_CASE_: Optional[int]     			=self._scattered(__lowerCAmelCase       ,		[n_queries, n_docs]       ,		target_type=torch.intaa		)
     SCREAMING_SNAKE_CASE_: Dict     			=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		)








"""simple docstring"""
def      __magic_name__				(							lowercase							,			lowercase ):
   return int((input_a, input_a).count(0 ) == 0 )





def      __magic_name__				(							):
   assert and_gate(0							,			0 ) == 0
   assert and_gate(0							,			1 ) == 0
   assert and_gate(1							,			0 ) == 0
   assert and_gate(1							,			1 ) == 1


if __name__ == "__main__":
   test_and_gate()
   print(and_gate(1, 0))
   print(and_gate(0, 0))
   print(and_gate(0, 1))
   print(and_gate(1, 1))



