


'''simple docstring'''






import os
import tempfile
import unittest

from transformers import is_torch_available
from transformers.testing_utils import require_torch


if is_torch_available():
						import torch
						from torch import nn

						from transformers import (
						    Adafactor,
						    AdamW,
						    get_constant_schedule,
						    get_constant_schedule_with_warmup,
						    get_cosine_schedule_with_warmup,
						    get_cosine_with_hard_restarts_schedule_with_warmup,
						    get_inverse_sqrt_schedule,
						    get_linear_schedule_with_warmup,
						    get_polynomial_decay_schedule_with_warmup,
						)


def       _snake_case     (  A				,					A=10 )		->					Optional[Any]:
							lowerCAmelCase__      =			[]
							for _ in range(__snake_case ):
														lrs.append(scheduler.get_lr()[0] )
														scheduler.step()
							return lrs


def       _snake_case     (  A				,					A=10 )		->					str:
							lowerCAmelCase__      =			[]
							for step in range(__snake_case ):
														lrs.append(scheduler.get_lr()[0] )
														scheduler.step()
														if step == num_steps // 2:
																					with tempfile.TemporaryDirectory() as tmpdirname:
																												lowerCAmelCase__      =			os.path.join(__snake_case				,					'''schedule.bin''' )
																												torch.save(scheduler.state_dict()				,					__snake_case )

																												lowerCAmelCase__      =			torch.load(__snake_case )
																												scheduler.load_state_dict(__snake_case )
							return lrs




@require_torch
class 	a__       (							unittest.TestCase						):





							'''simple docstring'''






							def 						__SCREAMING_SNAKE_CASE							(	self      ,					lowerCamelCase_      ,					lowerCamelCase_      ,					lowerCamelCase_     ) ->							List[str]:
														self.assertEqual(len(lowerCamelCase_     )      ,					len(lowerCamelCase_     )     )
														for a, b in zip(lowerCamelCase_      ,					lowerCamelCase_     ):
																					self.assertAlmostEqual(lowerCamelCase_      ,					lowerCamelCase_      ,					delta=lowerCamelCase_     )







							def 						__SCREAMING_SNAKE_CASE							(	self     ) ->							str:
														lowerCAmelCase__      =			torch.tensor([0.1, -0.2, -0.1]      ,					requires_grad=lowerCamelCase_     )
														lowerCAmelCase__      =			torch.tensor([0.4, 0.2, -0.5]     )
														lowerCAmelCase__      =			nn.MSELoss()
														# No warmup, constant schedule, no gradient clipping
														lowerCAmelCase__      =			AdamW(params=[w]      ,					lr=2e-1      ,					weight_decay=0.0     )
														for _ in range(1_00     ):
																					lowerCAmelCase__      =			criterion(lowerCamelCase_      ,					lowerCamelCase_     )
																					loss.backward()
																					optimizer.step()
																					w.grad.detach_()  # No zero_grad() function on simple tensors. we do it ourselves.
																					w.grad.zero_()
														self.assertListAlmostEqual(w.tolist()      ,					[0.4, 0.2, -0.5]      ,					tol=1e-2     )







							def 						__SCREAMING_SNAKE_CASE							(	self     ) ->							int:
														lowerCAmelCase__      =			torch.tensor([0.1, -0.2, -0.1]      ,					requires_grad=lowerCamelCase_     )
														lowerCAmelCase__      =			torch.tensor([0.4, 0.2, -0.5]     )
														lowerCAmelCase__      =			nn.MSELoss()
														# No warmup, constant schedule, no gradient clipping
														lowerCAmelCase__      =			Adafactor(
														    params=[w]      ,					lr=1e-2      ,					eps=(1e-30, 1e-3)      ,					clip_threshold=1.0      ,					decay_rate=-0.8      ,					betaa=lowerCamelCase_      ,					weight_decay=0.0      ,					relative_step=lowerCamelCase_      ,					scale_parameter=lowerCamelCase_      ,					warmup_init=lowerCamelCase_      ,					)
														for _ in range(10_00     ):
																					lowerCAmelCase__      =			criterion(lowerCamelCase_      ,					lowerCamelCase_     )
																					loss.backward()
																					optimizer.step()
																					w.grad.detach_()  # No zero_grad() function on simple tensors. we do it ourselves.
																					w.grad.zero_()
														self.assertListAlmostEqual(w.tolist()      ,					[0.4, 0.2, -0.5]      ,					tol=1e-2     )




@require_torch
class 	a__       (							unittest.TestCase						):





							'''simple docstring'''






							lowercase__    :     List[str]					=   nn.Linear(5_0       ,       5_0						) if is_torch_available() else None
							lowercase__    :     List[Any]					=   AdamW(m.parameters()       ,       lr=1_0.0						) if is_torch_available() else None
							lowercase__    :     Optional[Any]					=   1_0







							def 						__SCREAMING_SNAKE_CASE							(	self      ,					lowerCamelCase_      ,					lowerCamelCase_      ,					lowerCamelCase_      ,					lowerCamelCase_=None     ) ->							int:
														self.assertEqual(len(lowerCamelCase_     )      ,					len(lowerCamelCase_     )     )
														for a, b in zip(lowerCamelCase_      ,					lowerCamelCase_     ):
																					self.assertAlmostEqual(lowerCamelCase_      ,					lowerCamelCase_      ,					delta=lowerCamelCase_      ,					msg=lowerCamelCase_     )







							def 						__SCREAMING_SNAKE_CASE							(	self     ) ->							Optional[Any]:
														lowerCAmelCase__      =			{"num_warmup_steps": 2, "num_training_steps": 10}
														# schedulers doct format
														# function: (sched_args_dict, expected_learning_rates)
														lowerCAmelCase__      =			{
														    get_constant_schedule: ({}, [10.0] * self.num_steps),
														    get_constant_schedule_with_warmup: (
														        {"num_warmup_steps": 4},
														        [0.0, 2.5, 5.0, 7.5, 10.0, 10.0, 10.0, 10.0, 10.0, 10.0],
														    ),
														    get_linear_schedule_with_warmup: (
														        {**common_kwargs},
														        [0.0, 5.0, 10.0, 8.75, 7.5, 6.25, 5.0, 3.75, 2.5, 1.25],
														    ),
														    get_cosine_schedule_with_warmup: (
														        {**common_kwargs},
														        [0.0, 5.0, 10.0, 9.61, 8.53, 6.91, 5.0, 3.08, 1.46, 0.38],
														    ),
														    get_cosine_with_hard_restarts_schedule_with_warmup: (
														        {**common_kwargs, "num_cycles": 2},
														        [0.0, 5.0, 10.0, 8.53, 5.0, 1.46, 10.0, 8.53, 5.0, 1.46],
														    ),
														    get_polynomial_decay_schedule_with_warmup: (
														        {**common_kwargs, "power": 2.0, "lr_end": 1e-7},
														        [0.0, 5.0, 10.0, 7.656, 5.625, 3.906, 2.5, 1.406, 0.625, 0.156],
														    ),
														    get_inverse_sqrt_schedule: (
														        {"num_warmup_steps": 2},
														        [0.0, 5.0, 10.0, 8.165, 7.071, 6.325, 5.774, 5.345, 5.0, 4.714],
														    ),
														}

														for scheduler_func, data in scheds.items():
																					lowerCAmelCase__      =			data

																					lowerCAmelCase__      =			scheduler_func(self.optimizer      ,					**lowerCamelCase_     )
																					self.assertEqual(len([scheduler.get_lr()[0]]     )      ,					1     )
																					lowerCAmelCase__      =			unwrap_schedule(lowerCamelCase_      ,					self.num_steps     )
																					self.assertListAlmostEqual(
																					    lowerCamelCase_      ,					lowerCamelCase_      ,					tol=1e-2      ,					msg=F"""failed for {scheduler_func} in normal scheduler"""      ,					)

																					lowerCAmelCase__      =			scheduler_func(self.optimizer      ,					**lowerCamelCase_     )
																					if scheduler_func.__name__ != "get_constant_schedule":
																												LambdaScheduleWrapper.wrap_scheduler(lowerCamelCase_     )  # wrap to test picklability of the schedule
																					lowerCAmelCase__      =			unwrap_and_save_reload_schedule(lowerCamelCase_      ,					self.num_steps     )
																					self.assertListEqual(lowerCamelCase_      ,					lowerCamelCase_      ,					msg=F"""failed for {scheduler_func} in save and reload"""     )




class 	a__       :





							'''simple docstring'''






							def __init__(	self      ,					lowerCamelCase_     ) ->							List[str]:
														lowerCAmelCase__      =			fn







							def __call__(	self      ,					*lowerCamelCase_      ,					**lowerCamelCase_     ) ->							Dict:
														return self.fn(*lowerCamelCase_      ,					**lowerCamelCase_     )







							@classmethod
							def 						__SCREAMING_SNAKE_CASE							(	self      ,					lowerCamelCase_     ) ->							Any:
														lowerCAmelCase__      =			list(map(self      ,					scheduler.lr_lambdas     )     )



'''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__       (							a__						):





							'''simple docstring'''






							def __init__(	self      ,					lowerCamelCase_      ,					lowerCamelCase_      ,					lowerCamelCase_      ,					lowerCamelCase_=None     ) ->							Dict:
														super().__init__(
														    lowerCamelCase_      ,					question_encoder_tokenizer=lowerCamelCase_      ,					generator_tokenizer=lowerCamelCase_      ,					index=lowerCamelCase_      ,					init_retrieval=lowerCamelCase_      ,					)
														lowerCAmelCase__      =			None







							def 						__SCREAMING_SNAKE_CASE							(	self      ,					lowerCamelCase_     ) ->							str:
														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=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 						__SCREAMING_SNAKE_CASE							(	self     ) ->							Optional[Any]:
														return dist.get_rank(group=self.process_group     ) == 0







							def 						__SCREAMING_SNAKE_CASE							(	self      ,					lowerCamelCase_      ,					lowerCamelCase_      ,					lowerCamelCase_=torch.floataa     ) ->							Union[str, Any]:
														lowerCAmelCase__      =			torch.empty(lowerCamelCase_      ,					dtype=lowerCamelCase_     )
														dist.scatter(lowerCamelCase_      ,					src=0      ,					scatter_list=lowerCamelCase_      ,					group=self.process_group     )
														return target_tensor







							def 						__SCREAMING_SNAKE_CASE							(	self     ) ->							str:
														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'''     ))      ,					lowerCamelCase_     )
														return ifname







							def 						__SCREAMING_SNAKE_CASE							(	self      ,					lowerCamelCase_      ,					lowerCamelCase_     ) ->							Tuple[np.ndarray, List[dict]]:

														# single GPU training
														if not dist.is_initialized():
																					lowerCAmelCase__     ,      lowerCAmelCase__      =			self._main_retrieve(lowerCamelCase_      ,					lowerCamelCase_     )
																					return retrieved_doc_embeds, doc_ids, self.index.get_doc_dicts(lowerCamelCase_     )

														# 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(lowerCamelCase_     )]
														dist.gather(torch.tensor(lowerCamelCase_     )      ,					dst=0      ,					gather_list=lowerCamelCase_      ,					group=self.process_group     )

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