






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






						def __init__(		self  :Tuple			, __magic_name__  :str			, __magic_name__  :List[Any]			, __magic_name__  :Dict			, __magic_name__  :int=None				)      ->Optional[Any]:
										super().__init__(
										    __magic_name__			, question_encoder_tokenizer=__magic_name__			, generator_tokenizer=__magic_name__			, index=__magic_name__			, init_retrieval=__magic_name__			, )
										lowercase    :      str       = None






						def      __snake_case      (		self  :Dict			, __magic_name__  :int				)      ->Union[str, Any]:
										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    :      List[str]       = self._infer_socket_ifname()
														# avoid clash with the NCCL port
														lowercase    :      List[str]       = str(distributed_port + 1				)
														lowercase    :      str       = dist.new_group(ranks=__magic_name__			, 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      __snake_case      (		self  :List[Any]				)      ->Optional[int]:
										return dist.get_rank(group=self.process_group				) == 0






						def      __snake_case      (		self  :Tuple			, __magic_name__  :List[Any]			, __magic_name__  :Dict			, __magic_name__  :Dict=torch.floataa				)      ->List[str]:
										lowercase    :      int       = torch.empty(__magic_name__			, dtype=__magic_name__				)
										dist.scatter(__magic_name__			, src=0			, scatter_list=__magic_name__			, group=self.process_group				)
										return target_tensor






						def      __snake_case      (		self  :Optional[Any]				)      ->List[str]:
										lowercase    :      List[str]       = psutil.net_if_addrs()
										# a hacky way to deal with varying network interface names
										lowercase    :      Tuple       = next((addr for addr in addrs if addr.startswith("""e"""				))			, __magic_name__				)
										return ifname






						def      __snake_case      (		self  :Optional[Any]			, __magic_name__  :np.ndarray			, __magic_name__  :int				)      ->Tuple[np.ndarray, List[dict]]:

										# single GPU training
										if not dist.is_initialized():
														lowercase      ,      lowercase    :      Optional[Any]       = self._main_retrieve(__magic_name__			, __magic_name__				)
														return retrieved_doc_embeds, doc_ids, self.index.get_doc_dicts(__magic_name__				)

										# distributed training
										lowercase    :      List[str]       = dist.get_world_size(group=self.process_group				)

										# gather logic
										lowercase    :      List[str]       = None
										if self._is_main():
														lowercase    :      int       = [torch.empty(question_hidden_states.shape			, dtype=torch.floataa				) for _ in range(__magic_name__				)]
										dist.gather(torch.tensor(__magic_name__				)			, dst=0			, gather_list=__magic_name__			, group=self.process_group				)

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

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








"""simple docstring"""







import argparse
import logging
import pickle
from collections import Counter


logging.basicConfig(
    format='%(asctime)s - %(levelname)s - %(name)s - %(message)s', datefmt='%m/%d/%Y %H:%M:%S', level=logging.INFO
)
_lowerCAmelCase            =	logging.getLogger(__name__)

if __name__ == "__main__":
				_lowerCAmelCase            =	argparse.ArgumentParser(
				    description='Token Counts for smoothing the masking probabilities in MLM (cf XLM/word2vec)'
				)
				parser.add_argument(
				    '--data_file', type=str, default='data/dump.bert-base-uncased.pickle', help='The binarized dataset.'
				)
				parser.add_argument(
				    '--token_counts_dump', type=str, default='data/token_counts.bert-base-uncased.pickle', help='The dump file.'
				)
				parser.add_argument('--vocab_size', default=3_05_22, type=int)
				_lowerCAmelCase            =	parser.parse_args()

				logger.info(F'Loading data from {args.data_file}')
				with open(args.data_file, 'rb') as fp:
								_lowerCAmelCase            =	pickle.load(fp)

				logger.info('Counting occurrences for MLM.')
				_lowerCAmelCase            =	Counter()
				for tk_ids in data:
								counter.update(tk_ids)
				_lowerCAmelCase            =	[0] * args.vocab_size
				for k, v in counter.items():
								_lowerCAmelCase            =	v

				logger.info(F'Dump to {args.token_counts_dump}')
				with open(args.token_counts_dump, 'wb') as handle:
								pickle.dump(counts, handle, protocol=pickle.HIGHEST_PROTOCOL)
