



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

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




class 							lowerCAmelCase__  :







					def __init__(       self      :     List[Any]			,		_A      :     List[Any]):
											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
																	A__	:							Dict						    = deepcopy(_A)
											elif os.path.exists(_A):
																	with io.open(_A			,		"r"			,		encoding="utf-8") as f:
																							A__	:							List[str]						    = json.load(_A)
											else:
																	try:
																							A__	:							int						    = baseaa.urlsafe_baadecode(_A).decode("utf-8")
																							A__	:							Any						    = 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}')

											A__	:							List[Any]						    = config

											self.set_stage_and_offload()







					def  _lowercase			(       self      :     List[str]):
											# 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.
											A__	:							str						    = self.get_value("zero_optimization.stage"			,		-1)

											# offload
											A__	:							str						    = False
											if self.is_zeroa() or self.is_zeroa():
																	A__	:							str						    = set(["cpu", "nvme"])
																	A__	:							Tuple						    = 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:
																							A__	:							Union[str, Any]						    = True







					def  _lowercase			(       self      :     int			,		_A      :     Any):
											A__	:							Union[str, Any]						    = self.config

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

											return config, ds_key







					def  _lowercase			(       self      :     Tuple			,		_A      :     Optional[int]			,		_A      :     Dict=None):
											A__    ,       A__	:							str						    = self.find_config_node(_A)
											if config is None:
																	return default
											return config.get(_A			,		_A)







					def  _lowercase			(       self      :     Any			,		_A      :     str			,		_A      :     str=False):
											A__	:							int						    = self.config

											# find the config node of interest if it exists
											A__	:							Dict						    = ds_key_long.split(".")
											for node in nodes:
																	A__	:							List[str]						    = config
																	A__	:							Optional[int]						    = 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  _lowercase			(       self      :     Tuple			,		_A      :     Optional[Any]):
											A__	:							Any						    = self.get_value(_A)
											return False if value is None else bool(_A)







					def  _lowercase			(       self      :     Optional[Any]			,		_A      :     List[Any]):
											A__	:							Any						    = self.get_value(_A)
											return False if value is None else not bool(_A)







					def  _lowercase			(       self      :     Optional[int]):
											return self._stage == 2







					def  _lowercase			(       self      :     List[str]):
											return self._stage == 3







					def  _lowercase			(       self      :     Any):
											return self._offload




class 							lowerCAmelCase__  :







					def __init__(       self      :     Any			,		_A      :     Any):
											A__	:							int						    = engine







					def  _lowercase			(       self      :     List[str]			,		_A      :     List[str]			,		**_A      :     Any):
											# 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 							lowerCAmelCase__  (	UpperCamelCase						):







					def __init__(       self      :     Tuple			,		_A      :     int):
											super().__init__(_A			,		device_placement=_A			,		scaler=_A)
											A__	:							List[str]						    = hasattr(self.optimizer			,		"overflow")







					def  _lowercase			(       self      :     str			,		_A      :     int=None):
											pass  # `accelerator.backward(loss)` is doing that automatically. Therefore, its implementation is not needed







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







					@property
					def  _lowercase			(       self      :     Any):
											if self.__has_overflow__:
																	return self.optimizer.overflow
											return False




class 							lowerCAmelCase__  (	UpperCamelCase						):







					def __init__(       self      :     Optional[int]			,		_A      :     List[str]			,		_A      :     List[str]):
											super().__init__(_A			,		_A)







					def  _lowercase			(       self      :     Union[str, Any]):
											pass  # `accelerator.backward(loss)` is doing that automatically. Therefore, its implementation is not needed




class 							lowerCAmelCase__  :







					def __init__(       self      :     str			,		_A      :     List[Any]			,		_A      :     Optional[Any]=0.0_01			,		_A      :     List[str]=0			,		**_A      :     Optional[Any]):
											A__	:							List[Any]						    = params
											A__	:							Optional[Any]						    = lr
											A__	:							str						    = weight_decay
											A__	:							List[Any]						    = kwargs




class 							lowerCAmelCase__  :







					def __init__(       self      :     Optional[int]			,		_A      :     List[Any]			,		_A      :     Tuple=None			,		_A      :     Any=0			,		**_A      :     Dict):
											A__	:							int						    = optimizer
											A__	:							Tuple						    = total_num_steps
											A__	:							List[Any]						    = warmup_num_steps
											A__	:							Tuple						    = kwargs




import os
import sys
from contextlib import contextmanager


# Windows only
if os.name == "nt":
		import ctypes
		import msvcrt  # noqa




		class 							lowerCAmelCase__  (	ctypes.Structure						):
							# _fields is a specific attr expected by ctypes
							__A     :    Optional[Any]           =						[('size', ctypes.c_int), ('visible', ctypes.c_byte)]
def 	snake_case__       (					)		->		List[Any]:







						"""simple docstring"""

						if os.name == "nt":
												A__	:							Optional[Any]						    = CursorInfo()
												A__	:							Tuple						    = ctypes.windll.kernelaa.GetStdHandle(-1_1       )
												ctypes.windll.kernelaa.GetConsoleCursorInfo(__lowercase							,     ctypes.byref(__lowercase       )       )
												A__	:							Any						    = False
												ctypes.windll.kernelaa.SetConsoleCursorInfo(__lowercase							,     ctypes.byref(__lowercase       )       )
						elif os.name == "posix":
												sys.stdout.write("\033[?25l"       )
												sys.stdout.flush()
def 	snake_case__       (					)		->		Dict:







						"""simple docstring"""

						if os.name == "nt":
												A__	:							List[str]						    = CursorInfo()
												A__	:							Optional[int]						    = ctypes.windll.kernelaa.GetStdHandle(-1_1       )
												ctypes.windll.kernelaa.GetConsoleCursorInfo(__lowercase							,     ctypes.byref(__lowercase       )       )
												A__	:							int						    = True
												ctypes.windll.kernelaa.SetConsoleCursorInfo(__lowercase							,     ctypes.byref(__lowercase       )       )
						elif os.name == "posix":
												sys.stdout.write("\033[?25h"       )
												sys.stdout.flush()




@contextmanager
def 	snake_case__       (					)		->		Optional[int]:







						"""simple docstring"""

						try:
												hide_cursor()
												yield
						finally:
												show_cursor()