
"""simple docstring"""





import inspect
import warnings

import torch

from .state import AcceleratorState, GradientState
from .utils import DistributedType, honor_type, is_tpu_available


if is_tpu_available(check_device=False):
    import torch_xla.core.xla_model as xm







def       __a	(						__lowerCamelCase,						__lowerCamelCase    ):
   if isinstance(__lowerCamelCase,						(list, tuple)    ):
      return honor_type(__lowerCamelCase,						(move_to_device(__lowerCamelCase,						__lowerCamelCase    ) for t in state)    )
   elif isinstance(__lowerCamelCase,						__lowerCamelCase    ):
      return type(__lowerCamelCase    )({k: move_to_device(__lowerCamelCase,						__lowerCamelCase    ) for k, v in state.items()}    )
   elif isinstance(__lowerCamelCase,						torch.Tensor    ):
      return state.to(__lowerCamelCase    )
   return state


class A_      (torch.optim.Optimizer ):




  '''simple docstring'''







  def __init__(						self							,  lowercase_							,  lowercase_=True							,  lowercase_=None    ):

     """simple docstring"""
     UpperCAmelCase_   :   List[Any]     	=       optimizer
     UpperCAmelCase_   :   Any     	=       scaler
     UpperCAmelCase_   :   str     	=       AcceleratorState()
     UpperCAmelCase_   :   Tuple     	=       GradientState()
     UpperCAmelCase_   :   Dict     	=       device_placement
     UpperCAmelCase_   :   List[str]     	=       False
     UpperCAmelCase_   :   List[str]     	=       None

     # Handle device placement
     if device_placement:
        UpperCAmelCase_   :   Union[str, Any]     	=       self.optimizer.state_dict()
        if self.accelerator_state.distributed_type == DistributedType.TPU:
           xm.send_cpu_data_to_device(lowercase_							,  self.accelerator_state.device    )
        else:
           UpperCAmelCase_   :   Optional[int]     	=       move_to_device(lowercase_							,  self.accelerator_state.device    )
        self.optimizer.load_state_dict(lowercase_    )







  @property
  def 			UpperCamelCase__   (						self    ):

     """simple docstring"""
     return self.optimizer.state







  @state.setter
  def 			UpperCamelCase__   (						self							,  lowercase_    ):

     """simple docstring"""
     UpperCAmelCase_   :   Optional[Any]     	=       state







  @property
  def 			UpperCamelCase__   (						self    ):

     """simple docstring"""
     return self.optimizer.param_groups







  @param_groups.setter
  def 			UpperCamelCase__   (						self							,  lowercase_    ):

     """simple docstring"""
     UpperCAmelCase_   :   List[Any]     	=       param_groups







  @property
  def 			UpperCamelCase__   (						self    ):

     """simple docstring"""
     return self.optimizer.defaults







  @defaults.setter
  def 			UpperCamelCase__   (						self							,  lowercase_    ):

     """simple docstring"""
     UpperCAmelCase_   :   Dict     	=       defaults







  def 			UpperCamelCase__   (						self							,  lowercase_    ):

     """simple docstring"""
     self.optimizer.add_param_group(lowercase_    )







  def 			UpperCamelCase__   (						self							,  lowercase_    ):

     """simple docstring"""
     if self.accelerator_state.distributed_type == DistributedType.TPU and self.device_placement:
        xm.send_cpu_data_to_device(lowercase_							,  self.accelerator_state.device    )
     self.optimizer.load_state_dict(lowercase_    )







  def 			UpperCamelCase__   (						self    ):

     """simple docstring"""
     return self.optimizer.state_dict()







  def 			UpperCamelCase__   (						self							,  lowercase_=None    ):

     """simple docstring"""
     if self.gradient_state.sync_gradients:
        UpperCAmelCase_   :   int     	=       "set_to_none" in inspect.signature(self.optimizer.zero_grad    ).parameters
        if accept_arg:
           if set_to_none is None:
              UpperCAmelCase_   :   Optional[int]     	=       False
           self.optimizer.zero_grad(set_to_none=lowercase_    )
        else:
           if set_to_none is not None:
              raise ValueError("`set_to_none` for Optimizer.zero_grad` is not supported by this optimizer."    )
           self.optimizer.zero_grad()







  def 			UpperCamelCase__   (						self							,  lowercase_=None    ):

     """simple docstring"""
     if self.gradient_state.sync_gradients:
        if self.accelerator_state.distributed_type == DistributedType.TPU:
           UpperCAmelCase_   :   Any     	=       {"closure": closure} if closure is not None else {}
           xm.optimizer_step(self.optimizer							,  optimizer_args=lowercase_    )
        elif self.scaler is not None:
           UpperCAmelCase_   :   Tuple     	=       False
           if self._last_scale is None:
              # `get_scale` is an async operation requiring full synchronization
              # on CPU and GPUs before finishing. As a result, we store away
              # the prior one to reduce the call overhead
              UpperCAmelCase_   :   Any     	=       self.scaler.get_scale()
              UpperCAmelCase_   :   int     	=       True
           self.scaler.step(self.optimizer							,  lowercase_    )
           self.scaler.update()
           UpperCAmelCase_   :   int     	=       self.scaler.get_scale()
           if not new_scale:
              # If we reduced the loss scale, it means the optimizer step was skipped because of gradient overflow.
              UpperCAmelCase_   :   Union[str, Any]     	=       scale_after < self._last_scale
           UpperCAmelCase_   :   Union[str, Any]     	=       scale_after

        else:
           self.optimizer.step(lowercase_    )







  def 			UpperCamelCase__   (						self							,  lowercase_    ):

     """simple docstring"""
     for param_group in self.optimizer.param_groups:
        UpperCAmelCase_   :   List[Any]     	=       [parameters_map.get(lowercase_							,  lowercase_    ) for p in param_group["params"]]







  @property
  def 			UpperCamelCase__   (						self    ):

     """simple docstring"""
     warnings.warn(
         "The `is_overflow` property is deprecated and will be removed in version 1.0 of Accelerate use "
         "`optimizer.step_was_skipped` instead."							,  lowercase_							,  )
     return self._is_overflow







  @property
  def 			UpperCamelCase__   (						self    ):

     """simple docstring"""
     return self._is_overflow







  def __getstate__(						self    ):

     """simple docstring"""
     return self.__dict__.copy()







  def __setstate__(						self							,  lowercase_    ):

     """simple docstring"""
     self.__dict__.update(lowercase_    )









'''simple docstring'''


import logging
import random

import ray

from transformers import RagConfig, RagRetriever, RagTokenizer
from transformers.models.rag.retrieval_rag import CustomHFIndex


_lowerCamelCase					:						List[str]					     =							logging.getLogger(__name__)



class 				__UpperCAmelCase	:







		'''simple docstring'''







		def __init__(self    :   Any   ):
			A					=   False







		def 							A  (self    :   Tuple   ,   _lowerCAmelCase    :   Dict   ,   _lowerCAmelCase    :   Any   ,   _lowerCAmelCase    :   Tuple   ,   _lowerCAmelCase    :   Dict   ):
			if not self.initialized:
				A					=   RagRetriever(
				    _lowerCAmelCase   ,   question_encoder_tokenizer=_lowerCAmelCase   ,   generator_tokenizer=_lowerCAmelCase   ,   index=_lowerCAmelCase   ,   init_retrieval=_lowerCAmelCase   ,   )
				A					=   True







		def 							A  (self    :   List[str]   ):
			self.retriever.index.init_index()







		def 							A  (self    :   List[str]   ):
			# delete the old self.retriever object before assigning the new index
			del self.retriever
			A					=   False







		def 							A  (self    :   Any   ,   _lowerCAmelCase    :   List[str]   ,   _lowerCAmelCase    :   Optional[int]   ):
			A	,     A					=   self.retriever._main_retrieve(_lowerCAmelCase   ,   _lowerCAmelCase   )
			A					=   self.retriever.index.get_doc_dicts(_lowerCAmelCase   )
			return doc_ids, retrieved_doc_embeds, doc_dicts






class 				__UpperCAmelCase	(							A__    ):







		'''simple docstring'''







		def __init__(self    :   Any   ,   _lowerCAmelCase    :   Dict   ,   _lowerCAmelCase    :   Dict   ,   _lowerCAmelCase    :   Union[str, Any]   ,   _lowerCAmelCase    :   Dict   ,   _lowerCAmelCase    :   List[Any]=None   ):
			if index is not None and index.is_initialized() and len(_lowerCAmelCase   ) > 0:
				raise ValueError(
				    """When using Ray for distributed fine-tuning, """
				    """you'll need to provide the paths instead, """
				    """as the dataset and the index are loaded """
				    """separately. More info in examples/rag/use_own_knowledge_dataset.py """   )

			super().__init__(
			    _lowerCAmelCase   ,   question_encoder_tokenizer=_lowerCAmelCase   ,   generator_tokenizer=_lowerCAmelCase   ,   index=_lowerCAmelCase   ,   init_retrieval=_lowerCAmelCase   ,   )

			A					=   retrieval_workers
			A					=   question_encoder_tokenizer
			A					=   generator_tokenizer
			if len(self.retrieval_workers   ) > 0:
				ray.get(
				    [
				        worker.create_rag_retriever.remote(_lowerCAmelCase   ,   _lowerCAmelCase   ,   _lowerCAmelCase   ,   _lowerCAmelCase   )
				        for worker in self.retrieval_workers
				    ]   )







		def 							A  (self    :   Optional[Any]   ):
			logger.info("""initializing retrieval"""   )

			if len(self.retrieval_workers   ) > 0:
				ray.get([worker.init_retrieval.remote() for worker in self.retrieval_workers]   )
			else:
				# Non-distributed training. Load index into this same process.
				self.index.init_index()







		def 							A  (self    :   int   ,   _lowerCAmelCase    :   Dict   ,   _lowerCAmelCase    :   List[Any]   ):
			if len(self.retrieval_workers   ) > 0:
				# Select a random retrieval actor.
				A					=   self.retrieval_workers[random.randint(0   ,   len(self.retrieval_workers   ) - 1   )]
				A	,     A	,     A					=   ray.get(
				    random_worker.retrieve.remote(_lowerCAmelCase   ,   _lowerCAmelCase   )   )
			else:
				A	,     A					=   self._main_retrieve(_lowerCAmelCase   ,   _lowerCAmelCase   )
				A					=   self.index.get_doc_dicts(_lowerCAmelCase   )
			return retrieved_doc_embeds, doc_ids, doc_dicts







		@classmethod
		def 							A  (cls    :   Dict   ,   _lowerCAmelCase    :   Tuple   ,   _lowerCAmelCase    :   str=None   ,   **_lowerCAmelCase    :   List[Any]   ):
			return super(_lowerCAmelCase   ,   cls   ).get_tokenizers(_lowerCAmelCase   ,   _lowerCAmelCase   ,   **_lowerCAmelCase   )







		@classmethod
		def 							A  (cls    :   List[str]   ,   _lowerCAmelCase    :   List[str]   ,   _lowerCAmelCase    :   List[Any]   ,   _lowerCAmelCase    :   Optional[int]=None   ,   **_lowerCAmelCase    :   Optional[int]   ):
			A					=   kwargs.pop("""config"""   ,   _lowerCAmelCase   ) or RagConfig.from_pretrained(_lowerCAmelCase   ,   **_lowerCAmelCase   )
			A					=   RagTokenizer.from_pretrained(_lowerCAmelCase   ,   config=_lowerCAmelCase   )
			A					=   rag_tokenizer.question_encoder
			A					=   rag_tokenizer.generator

			if indexed_dataset is not None:
				A					=   """custom"""
				A					=   CustomHFIndex(config.retrieval_vector_size   ,   _lowerCAmelCase   )
			else:
				A					=   cls._build_index(_lowerCAmelCase   )

			return cls(
			    _lowerCAmelCase   ,   question_encoder_tokenizer=_lowerCAmelCase   ,   generator_tokenizer=_lowerCAmelCase   ,   retrieval_workers=_lowerCAmelCase   ,   index=_lowerCAmelCase   ,   )







		def 							A  (self    :   str   ):
			logger.info("""re-loading the new dataset with embeddings"""   )
			# access from the training loop

			ray.get([worker.clear_object.remote() for worker in self.retrieval_workers]   )

			# build the index object again
			A					=   self._build_index(self.config   )

			ray.get(
			    [
			        worker.create_rag_retriever.remote(
			            self.config   ,   self.question_encoder_tokenizer   ,   self.generator_tokenizer   ,   _lowerCAmelCase   )
			        for worker in self.retrieval_workers
			    ]   )

