




'''simple docstring'''







import baseaa
import io
import json
import os
from copy import deepcopy

from ..optimizer import AcceleratedOptimizer
from ..scheduler import AcceleratedScheduler







class 					_a :





     '''simple docstring'''



     def __init__(				self							,__a							)      ->				List[str]:
          if isinstance(__a							,__a							):
               # Don't modify user's data should they want to reuse it (e.g. in tests), because once we
               # modified it, it will not be accepted here again, since `auto` values would have been overridden
               snake_case			:      List[Any]        =			deepcopy(__a							)
          elif os.path.exists(__a							):
               with io.open(__a							,"""r"""							,encoding="""utf-8"""							) as f:
                    snake_case			:      int        =			json.load(__a							)
          else:
               try:
                    snake_case			:      Any        =			baseaa.urlsafe_baadecode(__a							).decode("""utf-8"""							)
                    snake_case			:      Tuple        =			json.loads(__a							)
               except (UnicodeDecodeError, AttributeError, ValueError):
                    raise ValueError(
                        F'''Expected a string path to an existing deepspeed config, or a dictionary, or a base64 encoded string. Received: {config_file_or_dict}'''							)

          snake_case			:      Tuple        =			config

          self.set_stage_and_offload()







     def 			snake_case_      (				self							)      ->				Any:
          # zero stage - this is done as early as possible, before model is created, to allow
          # ``is_deepspeed_zero3_enabled`` query and getting to the early deepspeed config object
          # during ``zero.Init()`` which needs to know the dtype, and some other hparams.
          snake_case			:      int        =			self.get_value("""zero_optimization.stage"""							,-1							)

          # offload
          snake_case			:      Optional[int]        =			False
          if self.is_zeroa() or self.is_zeroa():
               snake_case			:      List[str]        =			set(["""cpu""", """nvme"""]							)
               snake_case			:      Union[str, Any]        =			set(
                   [
                       self.get_value("""zero_optimization.offload_optimizer.device"""							),
                       self.get_value("""zero_optimization.offload_param.device"""							),
                   ]							)
               if len(offload_devices & offload_devices_valid							) > 0:
                    snake_case			:      Tuple        =			True







     def 			snake_case_      (				self							,__a							)      ->				Union[str, Any]:
          snake_case			:      Tuple        =			self.config

          # find the config node of interest if it exists
          snake_case			:      Optional[int]        =			ds_key_long.split("""."""							)
          snake_case			:      List[Any]        =			nodes.pop()
          for node in nodes:
               snake_case			:      str        =			config.get(__a							)
               if config is None:
                    return None, ds_key

          return config, ds_key







     def 			snake_case_      (				self							,__a							,__a=None							)      ->				Union[str, Any]:
          snake_case ,		snake_case			:      Any        =			self.find_config_node(__a							)
          if config is None:
               return default
          return config.get(__a							,__a							)







     def 			snake_case_      (				self							,__a							,__a=False							)      ->				Tuple:
          snake_case			:      List[str]        =			self.config

          # find the config node of interest if it exists
          snake_case			:      Optional[int]        =			ds_key_long.split("""."""							)
          for node in nodes:
               snake_case			:      List[Any]        =			config
               snake_case			:      Union[str, Any]        =			config.get(__a							)
               if config is None:
                    if must_exist:
                         raise ValueError(F'''Can\'t find {ds_key_long} entry in the config: {self.config}'''							)
                    else:
                         return

        # if found remove it
          if parent_config is not None:
               parent_config.pop(__a							)







     def 			snake_case_      (				self							,__a							)      ->				Optional[int]:
          snake_case			:      Union[str, Any]        =			self.get_value(__a							)
          return False if value is None else bool(__a							)







     def 			snake_case_      (				self							,__a							)      ->				int:
          snake_case			:      Optional[Any]        =			self.get_value(__a							)
          return False if value is None else not bool(__a							)







     def 			snake_case_      (				self							)      ->				Union[str, Any]:
          return self._stage == 2







     def 			snake_case_      (				self							)      ->				Optional[Any]:
          return self._stage == 3







     def 			snake_case_      (				self							)      ->				List[Any]:
          return self._offload







class 					_a :





     '''simple docstring'''



     def __init__(				self							,__a							)      ->				int:
          snake_case			:      str        =			engine







     def 			snake_case_      (				self							,__a							,**__a							)      ->				List[str]:
          # runs backpropagation and handles mixed precision
          self.engine.backward(__a							,**__a							)

          # Deepspeed's `engine.step` performs the following operations:
          # - gradient accumulation check
          # - gradient clipping
          # - optimizer step
          # - zero grad
          # - checking overflow
          # - lr_scheduler step (only if engine.lr_scheduler is not None)
          self.engine.step()
          # and this plugin overrides the above calls with no-ops when Accelerate runs under
          # Deepspeed, but allows normal functionality for non-Deepspeed cases thus enabling a simple
          # training loop that works transparently under many training regimes.







class 					_a (a__	):





     '''simple docstring'''



     def __init__(				self							,__a							)      ->				Optional[Any]:
          super().__init__(__a							,device_placement=__a							,scaler=__a							)
          snake_case			:      List[Any]        =			hasattr(self.optimizer							,"""overflow"""							)







     def 			snake_case_      (				self							,__a=None							)      ->				Tuple:
          pass  # `accelerator.backward(loss)` is doing that automatically. Therefore, its implementation is not needed







     def 			snake_case_      (				self							)      ->				List[str]:
          pass  # `accelerator.backward(loss)` is doing that automatically. Therefore, its implementation is not needed







     @property
     def 			snake_case_      (				self							)      ->				Optional[Any]:
          if self.__has_overflow__:
               return self.optimizer.overflow
          return False







class 					_a (a__	):





     '''simple docstring'''



     def __init__(				self							,__a							,__a							)      ->				Any:
          super().__init__(__a							,__a							)







     def 			snake_case_      (				self							)      ->				Optional[Any]:
          pass  # `accelerator.backward(loss)` is doing that automatically. Therefore, its implementation is not needed







class 					_a :





     '''simple docstring'''



     def __init__(				self							,__a							,__a=0.001							,__a=0							,**__a							)      ->				Tuple:
          snake_case			:      Optional[Any]        =			params
          snake_case			:      Any        =			lr
          snake_case			:      str        =			weight_decay
          snake_case			:      Tuple        =			kwargs







class 					_a :





     '''simple docstring'''



     def __init__(				self							,__a							,__a=None							,__a=0							,**__a							)      ->				List[str]:
          snake_case			:      Optional[Any]        =			optimizer
          snake_case			:      Union[str, Any]        =			total_num_steps
          snake_case			:      int        =			warmup_num_steps
          snake_case			:      Union[str, Any]        =			kwargs









'''simple docstring'''







import os


# All paths are set with the intent you should run this script from the root of the repo with the command
# python utils/check_doctest_list.py
lowercase      :   List[str]      =       """."""


if __name__ == "__main__":
      lowercase      :   List[str]      =       os.path.join(REPO_PATH, """utils/documentation_tests.txt""")
      lowercase      :   Union[str, Any]      =       []
      lowercase      :   List[str]      =       []
      with open(doctest_file_path) as fp:
            for line in fp:
                  lowercase      :   List[str]      =       line.strip()
                  lowercase      :   Optional[Any]      =       os.path.join(REPO_PATH, line)
                  if not (os.path.isfile(path) or os.path.isdir(path)):
                        non_existent_paths.append(line)
                  all_paths.append(path)
      if len(non_existent_paths) > 0:
            lowercase      :   int      =       """\n""".join(non_existent_paths)
            raise ValueError(F"""`utils/documentation_tests.txt` contains non-existent paths:\n{non_existent_paths}""")
      if all_paths != sorted(all_paths):
            raise ValueError("""Files in `utils/documentation_tests.txt` are not in alphabetical order.""")



