
"""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 			lowercase					( lowercase__		):






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





				"""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_      ,)
				lowerCAmelCase												=  None






			def 					UpperCAmelCase							(self  :	Any      ,SCREAMING_SNAKE_CASE_  :	int			)       ->	List[str]:





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





				"""simple docstring"""


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






			def 					UpperCAmelCase							(self  :	List[str]      ,SCREAMING_SNAKE_CASE_  :	Dict      ,SCREAMING_SNAKE_CASE_  :	Optional[int]      ,SCREAMING_SNAKE_CASE_  :	List[str]=torch.floataa			)       ->	List[Any]:





				"""simple docstring"""


				lowerCAmelCase												=  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 					UpperCAmelCase							(self  :	Union[str, Any]			)       ->	Optional[Any]:





				"""simple docstring"""


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






			def 					UpperCAmelCase							(self  :	Union[str, Any]      ,SCREAMING_SNAKE_CASE_  :	np.ndarray      ,SCREAMING_SNAKE_CASE_  :	int			)       ->	Tuple[np.ndarray, List[dict]]:





				"""simple docstring"""


				if not dist.is_initialized():
					lowerCAmelCase							,       lowerCAmelCase												=  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
				lowerCAmelCase												=  dist.get_world_size(group=self.process_group			)

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




"""simple docstring"""






from typing import List, Optional, Union

from ...image_utils import ImageInput
from ...processing_utils import ProcessorMixin
from ...tokenization_utils_base import BatchEncoding, PaddingStrategy, PreTokenizedInput, TextInput, TruncationStrategy
from ...utils import TensorType



class 			lowercase					( lowercase__		):
			lowercase        =     ['''image_processor''', '''tokenizer''']
			lowercase        =     '''BlipImageProcessor'''
			lowercase        =     ('''BertTokenizer''', '''BertTokenizerFast''')






			def __init__(self  :	Any      ,SCREAMING_SNAKE_CASE_  :	List[str]      ,SCREAMING_SNAKE_CASE_  :	Optional[int]			)       ->	Dict:





				"""simple docstring"""


				lowerCAmelCase												=  False
				super().__init__(SCREAMING_SNAKE_CASE_      ,SCREAMING_SNAKE_CASE_			)
				lowerCAmelCase												=  self.image_processor






			def __call__(self  :	int      ,SCREAMING_SNAKE_CASE_  :	ImageInput = None      ,SCREAMING_SNAKE_CASE_  :	Union[TextInput, PreTokenizedInput, List[TextInput], List[PreTokenizedInput]] = None      ,SCREAMING_SNAKE_CASE_  :	bool = True      ,SCREAMING_SNAKE_CASE_  :	Union[bool, str, PaddingStrategy] = False      ,SCREAMING_SNAKE_CASE_  :	Union[bool, str, TruncationStrategy] = None      ,SCREAMING_SNAKE_CASE_  :	Optional[int] = None      ,SCREAMING_SNAKE_CASE_  :	int = 0      ,SCREAMING_SNAKE_CASE_  :	Optional[int] = None      ,SCREAMING_SNAKE_CASE_  :	Optional[bool] = None      ,SCREAMING_SNAKE_CASE_  :	bool = False      ,SCREAMING_SNAKE_CASE_  :	bool = False      ,SCREAMING_SNAKE_CASE_  :	bool = False      ,SCREAMING_SNAKE_CASE_  :	bool = False      ,SCREAMING_SNAKE_CASE_  :	bool = False      ,SCREAMING_SNAKE_CASE_  :	bool = True      ,SCREAMING_SNAKE_CASE_  :	Optional[Union[str, TensorType]] = None      ,**SCREAMING_SNAKE_CASE_  :	str      ,)       ->	BatchEncoding:





				"""simple docstring"""


				if images is None and text is None:
					raise ValueError('''You have to specify either images or text.'''			)

				# Get only text
				if images is None:
					lowerCAmelCase												=  self.tokenizer
					lowerCAmelCase												=  self.tokenizer(
					    text=SCREAMING_SNAKE_CASE_      ,add_special_tokens=SCREAMING_SNAKE_CASE_      ,padding=SCREAMING_SNAKE_CASE_      ,truncation=SCREAMING_SNAKE_CASE_      ,max_length=SCREAMING_SNAKE_CASE_      ,stride=SCREAMING_SNAKE_CASE_      ,pad_to_multiple_of=SCREAMING_SNAKE_CASE_      ,return_attention_mask=SCREAMING_SNAKE_CASE_      ,return_overflowing_tokens=SCREAMING_SNAKE_CASE_      ,return_special_tokens_mask=SCREAMING_SNAKE_CASE_      ,return_offsets_mapping=SCREAMING_SNAKE_CASE_      ,return_token_type_ids=SCREAMING_SNAKE_CASE_      ,return_length=SCREAMING_SNAKE_CASE_      ,verbose=SCREAMING_SNAKE_CASE_      ,return_tensors=SCREAMING_SNAKE_CASE_      ,**SCREAMING_SNAKE_CASE_      ,)
					return text_encoding

				# add pixel_values
				lowerCAmelCase												=  self.image_processor(SCREAMING_SNAKE_CASE_      ,return_tensors=SCREAMING_SNAKE_CASE_			)

				if text is not None:
					lowerCAmelCase												=  self.tokenizer(
					    text=SCREAMING_SNAKE_CASE_      ,add_special_tokens=SCREAMING_SNAKE_CASE_      ,padding=SCREAMING_SNAKE_CASE_      ,truncation=SCREAMING_SNAKE_CASE_      ,max_length=SCREAMING_SNAKE_CASE_      ,stride=SCREAMING_SNAKE_CASE_      ,pad_to_multiple_of=SCREAMING_SNAKE_CASE_      ,return_attention_mask=SCREAMING_SNAKE_CASE_      ,return_overflowing_tokens=SCREAMING_SNAKE_CASE_      ,return_special_tokens_mask=SCREAMING_SNAKE_CASE_      ,return_offsets_mapping=SCREAMING_SNAKE_CASE_      ,return_token_type_ids=SCREAMING_SNAKE_CASE_      ,return_length=SCREAMING_SNAKE_CASE_      ,verbose=SCREAMING_SNAKE_CASE_      ,return_tensors=SCREAMING_SNAKE_CASE_      ,**SCREAMING_SNAKE_CASE_      ,)
				else:
					lowerCAmelCase												=  None

				if text_encoding is not None:
					encoding_image_processor.update(SCREAMING_SNAKE_CASE_			)

				return encoding_image_processor






			def 					UpperCAmelCase							(self  :	List[str]      ,*SCREAMING_SNAKE_CASE_  :	str      ,**SCREAMING_SNAKE_CASE_  :	str			)       ->	Union[str, Any]:





				"""simple docstring"""


				return self.tokenizer.batch_decode(*SCREAMING_SNAKE_CASE_      ,**SCREAMING_SNAKE_CASE_			)






			def 					UpperCAmelCase							(self  :	List[Any]      ,*SCREAMING_SNAKE_CASE_  :	str      ,**SCREAMING_SNAKE_CASE_  :	Optional[int]			)       ->	Any:





				"""simple docstring"""


				return self.tokenizer.decode(*SCREAMING_SNAKE_CASE_      ,**SCREAMING_SNAKE_CASE_			)






			@property
			def 					UpperCAmelCase							(self  :	Dict			)       ->	Tuple:





				"""simple docstring"""


				lowerCAmelCase												=  self.tokenizer.model_input_names
				lowerCAmelCase												=  self.image_processor.model_input_names
				return list(dict.fromkeys(tokenizer_input_names + image_processor_input_names			)			)


