



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


lowerCamelCase		       =				logging.getLogger(__name__)





class _UpperCamelCase						(   UpperCAmelCase_						):





				'''simple docstring'''





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







											'''simple docstring'''

											super().__init__(
											    lowerCAmelCase__     ,				question_encoder_tokenizer=lowerCAmelCase__     ,				generator_tokenizer=lowerCAmelCase__     ,				index=lowerCAmelCase__     ,				init_retrieval=lowerCAmelCase__     ,				)
											__lowercase						=None

				def  __lowerCamelCase   (							self     :     List[Any]     ,				_lowerCAmelCase     :     int):







											'''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
																		__lowercase						=self._infer_socket_ifname()
																		# avoid clash with the NCCL port
																		__lowercase						=str(distributed_port + 1)
																		__lowercase						=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     :     Optional[int]):







											'''simple docstring'''

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

				def  __lowerCamelCase   (							self     :     Dict     ,				_lowerCAmelCase     :     Union[str, Any]     ,				_lowerCAmelCase     :     Optional[int]     ,				_lowerCAmelCase     :     Optional[int]=torch.floataa):







											'''simple docstring'''

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

				def  __lowerCamelCase   (							self     :     List[Any]):







											'''simple docstring'''

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

				def  __lowerCamelCase   (							self     :     str     ,				_lowerCAmelCase     :     np.ndarray     ,				_lowerCAmelCase     :     int):







											'''simple docstring'''

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

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

											# gather logic
											__lowercase						=None
											if self._is_main():
																		__lowercase						=[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
											__lowercase						=question_hidden_states.shape[0]
											__lowercase						=[]
											__lowercase						=[]
											if self._is_main():
																		assert len(lowerCAmelCase__) == world_size
																		__lowercase						=self._main_retrieve(torch.cat(lowerCAmelCase__).numpy()     ,				lowerCAmelCase__)
																		__lowercase						=torch.tensor(lowerCAmelCase__), torch.tensor(lowerCAmelCase__)
																		__lowercase						=self._chunk_tensor(lowerCAmelCase__     ,				lowerCAmelCase__)
																		__lowercase						=self._chunk_tensor(lowerCAmelCase__     ,				lowerCAmelCase__)
											__lowercase						=self._scattered(lowerCAmelCase__     ,				[n_queries, n_docs]     ,				target_type=torch.intaa)
											__lowercase						=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 collections.abc import Callable



class   __lowercase				:




			"""simple docstring"""







			def __init__( self  :     Tuple   ,						lowerCAmelCase__  :     Callable | None = None):
									# Stores actual heap items.
									SCREAMING_SNAKE_CASE_: list  		=       []
									# Stores indexes of each item for supporting updates and deletion.
									SCREAMING_SNAKE_CASE_: dict  		=       {}
									# Stores current size of heap.
									SCREAMING_SNAKE_CASE_: Optional[Any]  		=       0
									# Stores function used to evaluate the score of an item on which basis ordering
									# will be done.
									SCREAMING_SNAKE_CASE_: Any  		=       key or (lambda lowerCAmelCase__: x)



			def        _SCREAMING_SNAKE_CASE			( self  :     Dict   ,						lowerCAmelCase__  :     int):
									return int((i - 1) / 2) if i > 0 else None



			def        _SCREAMING_SNAKE_CASE			( self  :     Optional[Any]   ,						lowerCAmelCase__  :     int):
									SCREAMING_SNAKE_CASE_: Union[str, Any]  		=       int(2 * i + 1)
									return left if 0 < left < self.size else None



			def        _SCREAMING_SNAKE_CASE			( self  :     Union[str, Any]   ,						lowerCAmelCase__  :     int):
									SCREAMING_SNAKE_CASE_: Union[str, Any]  		=       int(2 * i + 2)
									return right if 0 < right < self.size else None



			def        _SCREAMING_SNAKE_CASE			( self  :     str   ,						lowerCAmelCase__  :     int   ,						lowerCAmelCase__  :     int):
									SCREAMING_SNAKE_CASE_		,    SCREAMING_SNAKE_CASE_: str  		=       (
									    self.pos_map[self.arr[j][0]],
									    self.pos_map[self.arr[i][0]],
									)
									# Then swap the items in the list.
									SCREAMING_SNAKE_CASE_		,    SCREAMING_SNAKE_CASE_: List[str]  		=       self.arr[j], self.arr[i]



			def        _SCREAMING_SNAKE_CASE			( self  :     str   ,						lowerCAmelCase__  :     int   ,						lowerCAmelCase__  :     int):
									return self.arr[i][1] < self.arr[j][1]



			def        _SCREAMING_SNAKE_CASE			( self  :     str   ,						lowerCAmelCase__  :     int):
									SCREAMING_SNAKE_CASE_: Any  		=       self._left(lowerCAmelCase__)
									SCREAMING_SNAKE_CASE_: List[Any]  		=       self._right(lowerCAmelCase__)
									SCREAMING_SNAKE_CASE_: List[Any]  		=       i

									if left is not None and not self._cmp(lowerCAmelCase__   ,						lowerCAmelCase__):
															SCREAMING_SNAKE_CASE_: Optional[int]  		=       left
									if right is not None and not self._cmp(lowerCAmelCase__   ,						lowerCAmelCase__):
															SCREAMING_SNAKE_CASE_: Tuple  		=       right

									return valid_parent



			def        _SCREAMING_SNAKE_CASE			( self  :     Optional[int]   ,						lowerCAmelCase__  :     int):
									SCREAMING_SNAKE_CASE_: List[Any]  		=       self._parent(lowerCAmelCase__)
									while parent is not None and not self._cmp(lowerCAmelCase__   ,						lowerCAmelCase__):
															self._swap(lowerCAmelCase__   ,						lowerCAmelCase__)
															SCREAMING_SNAKE_CASE_		,    SCREAMING_SNAKE_CASE_: List[str]  		=       parent, self._parent(lowerCAmelCase__)



			def        _SCREAMING_SNAKE_CASE			( self  :     Optional[Any]   ,						lowerCAmelCase__  :     int):
									SCREAMING_SNAKE_CASE_: Optional[int]  		=       self._get_valid_parent(lowerCAmelCase__)
									while valid_parent != index:
															self._swap(lowerCAmelCase__   ,						lowerCAmelCase__)
															SCREAMING_SNAKE_CASE_		,    SCREAMING_SNAKE_CASE_: Dict  		=       valid_parent, self._get_valid_parent(lowerCAmelCase__)



			def        _SCREAMING_SNAKE_CASE			( self  :     Optional[int]   ,						lowerCAmelCase__  :     int   ,						lowerCAmelCase__  :     int):
									if item not in self.pos_map:
															return
									SCREAMING_SNAKE_CASE_: Any  		=       self.pos_map[item]
									SCREAMING_SNAKE_CASE_: int  		=       [item, self.key(lowerCAmelCase__)]
									# Make sure heap is right in both up and down direction.
									# Ideally only one of them will make any change.
									self._heapify_up(lowerCAmelCase__)
									self._heapify_down(lowerCAmelCase__)



			def        _SCREAMING_SNAKE_CASE			( self  :     List[str]   ,						lowerCAmelCase__  :     int):
									if item not in self.pos_map:
															return
									SCREAMING_SNAKE_CASE_: Optional[Any]  		=       self.pos_map[item]
									del self.pos_map[item]
									SCREAMING_SNAKE_CASE_: List[str]  		=       self.arr[self.size - 1]
									SCREAMING_SNAKE_CASE_: Tuple  		=       index
									self.size -= 1
									# Make sure heap is right in both up and down direction. Ideally only one
									# of them will make any change- so no performance loss in calling both.
									if self.size > index:
															self._heapify_up(lowerCAmelCase__)
															self._heapify_down(lowerCAmelCase__)



			def        _SCREAMING_SNAKE_CASE			( self  :     Optional[Any]   ,						lowerCAmelCase__  :     int   ,						lowerCAmelCase__  :     int):
									SCREAMING_SNAKE_CASE_: Optional[int]  		=       len(self.arr)
									if arr_len == self.size:
															self.arr.append([item, self.key(lowerCAmelCase__)])
									else:
															SCREAMING_SNAKE_CASE_: str  		=       [item, self.key(lowerCAmelCase__)]
									SCREAMING_SNAKE_CASE_: List[Any]  		=       self.size
									self.size += 1
									self._heapify_up(self.size - 1)



			def        _SCREAMING_SNAKE_CASE			( self  :     List[Any]):
									return self.arr[0] if self.size else None




			def        _SCREAMING_SNAKE_CASE			( self  :     Dict):
									SCREAMING_SNAKE_CASE_: Dict  		=       self.get_top()
									if top_item_tuple:
															self.delete_item(top_item_tuple[0])
									return top_item_tuple







def 			A_      (    ):
						pass


if __name__ == "__main__":
						import doctest

						doctest.testmod()






