
'''simple docstring'''







from typing import List, Optional, TypeVar

from .arrow_dataset import Dataset, _concatenate_map_style_datasets, _interleave_map_style_datasets
from .dataset_dict import DatasetDict, IterableDatasetDict
from .info import DatasetInfo
from .iterable_dataset import IterableDataset, _concatenate_iterable_datasets, _interleave_iterable_datasets
from .splits import NamedSplit
from .utils import logging
from .utils.py_utils import Literal


_snake_case				:   Dict     =				logging.get_logger(__name__)


_snake_case				:   Tuple     =				TypeVar("""DatasetType""", Dataset, IterableDataset)



def 			_a		(	_SCREAMING_SNAKE_CASE		:   List[DatasetType]	,			_SCREAMING_SNAKE_CASE		:   Optional[List[float]] = None	,			_SCREAMING_SNAKE_CASE		:   Optional[int] = None	,			_SCREAMING_SNAKE_CASE		:   Optional[DatasetInfo] = None	,			_SCREAMING_SNAKE_CASE		:   Optional[NamedSplit] = None	,			_SCREAMING_SNAKE_CASE		:   Literal["first_exhausted", "all_exhausted"] = "first_exhausted"	,			):
						from .arrow_dataset import Dataset
						from .iterable_dataset import IterableDataset

						if not datasets:
												raise ValueError("Unable to interleave an empty list of datasets."  )
						for i, dataset in enumerate(_SCREAMING_SNAKE_CASE  ):
												if not isinstance(_SCREAMING_SNAKE_CASE	,			(Dataset, IterableDataset)  ):
																		if isinstance(_SCREAMING_SNAKE_CASE	,			(DatasetDict, IterableDatasetDict)  ):
																								if not dataset:
																														raise ValueError(
																														    F'Expected a list of Dataset objects or a list of IterableDataset objects, but element at position {i} '
																														    "is an empty dataset dictionary."  )
																								raise ValueError(
																								    F'Dataset at position {i} has at least one split: {list(_SCREAMING_SNAKE_CASE  )}\n'
																								    F'Please pick one to interleave with the other datasets, for example: dataset[\'{next(iter(_SCREAMING_SNAKE_CASE  )  )}\']'  )
																		raise ValueError(
																		    F'Expected a list of Dataset objects or a list of IterableDataset objects, but element at position {i} is a {type(_SCREAMING_SNAKE_CASE  ).__name__}.'  )
												if i == 0:
																		_SCREAMING_SNAKE_CASE,							_SCREAMING_SNAKE_CASE				= (
																		    (Dataset, IterableDataset) if isinstance(_SCREAMING_SNAKE_CASE	,			_SCREAMING_SNAKE_CASE  ) else (IterableDataset, Dataset)
																		)
												elif not isinstance(_SCREAMING_SNAKE_CASE	,			_SCREAMING_SNAKE_CASE  ):
																		raise ValueError(
																		    F'Unable to interleave a {dataset_type.__name__} (at position 0) with a {other_type.__name__} (at position {i}). Expected a list of Dataset objects or a list of IterableDataset objects.'  )
						if stopping_strategy not in ["first_exhausted", "all_exhausted"]:
												raise ValueError(F'{stopping_strategy} is not supported. Please enter a valid stopping_strategy.'  )
						if dataset_type is Dataset:
												return _interleave_map_style_datasets(
												    _SCREAMING_SNAKE_CASE	,			_SCREAMING_SNAKE_CASE	,			_SCREAMING_SNAKE_CASE	,			info=_SCREAMING_SNAKE_CASE	,			split=_SCREAMING_SNAKE_CASE	,			stopping_strategy=_SCREAMING_SNAKE_CASE  )
						else:
												return _interleave_iterable_datasets(
												    _SCREAMING_SNAKE_CASE	,			_SCREAMING_SNAKE_CASE	,			_SCREAMING_SNAKE_CASE	,			info=_SCREAMING_SNAKE_CASE	,			split=_SCREAMING_SNAKE_CASE	,			stopping_strategy=_SCREAMING_SNAKE_CASE  )




def 			_a		(	_SCREAMING_SNAKE_CASE		:   List[DatasetType]	,			_SCREAMING_SNAKE_CASE		:   Optional[DatasetInfo] = None	,			_SCREAMING_SNAKE_CASE		:   Optional[NamedSplit] = None	,			_SCREAMING_SNAKE_CASE		:   int = 0	,			):

						if not dsets:
												raise ValueError("Unable to concatenate an empty list of datasets."  )
						for i, dataset in enumerate(_SCREAMING_SNAKE_CASE  ):
												if not isinstance(_SCREAMING_SNAKE_CASE	,			(Dataset, IterableDataset)  ):
																		if isinstance(_SCREAMING_SNAKE_CASE	,			(DatasetDict, IterableDatasetDict)  ):
																								if not dataset:
																														raise ValueError(
																														    F'Expected a list of Dataset objects or a list of IterableDataset objects, but element at position {i} '
																														    "is an empty dataset dictionary."  )
																								raise ValueError(
																								    F'Dataset at position {i} has at least one split: {list(_SCREAMING_SNAKE_CASE  )}\n'
																								    F'Please pick one to interleave with the other datasets, for example: dataset[\'{next(iter(_SCREAMING_SNAKE_CASE  )  )}\']'  )
																		raise ValueError(
																		    F'Expected a list of Dataset objects or a list of IterableDataset objects, but element at position {i} is a {type(_SCREAMING_SNAKE_CASE  ).__name__}.'  )
												if i == 0:
																		_SCREAMING_SNAKE_CASE,							_SCREAMING_SNAKE_CASE				= (
																		    (Dataset, IterableDataset) if isinstance(_SCREAMING_SNAKE_CASE	,			_SCREAMING_SNAKE_CASE  ) else (IterableDataset, Dataset)
																		)
												elif not isinstance(_SCREAMING_SNAKE_CASE	,			_SCREAMING_SNAKE_CASE  ):
																		raise ValueError(
																		    F'Unable to interleave a {dataset_type.__name__} (at position 0) with a {other_type.__name__} (at position {i}). Expected a list of Dataset objects or a list of IterableDataset objects.'  )
						if dataset_type is Dataset:
												return _concatenate_map_style_datasets(_SCREAMING_SNAKE_CASE	,			info=_SCREAMING_SNAKE_CASE	,			split=_SCREAMING_SNAKE_CASE	,			axis=_SCREAMING_SNAKE_CASE  )
						else:
												return _concatenate_iterable_datasets(_SCREAMING_SNAKE_CASE	,			info=_SCREAMING_SNAKE_CASE	,			split=_SCREAMING_SNAKE_CASE	,			axis=_SCREAMING_SNAKE_CASE  )

'''simple docstring'''







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					,						UpperCamelCase      ):
								if isinstance(UpperCamelCase					,						UpperCamelCase      ):
														# 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
														_SCREAMING_SNAKE_CASE				= deepcopy(UpperCamelCase      )
								elif os.path.exists(UpperCamelCase      ):
														with io.open(UpperCamelCase					,						"r"					,						encoding="utf-8"      ) as f:
																				_SCREAMING_SNAKE_CASE				= json.load(UpperCamelCase      )
								else:
														try:
																				_SCREAMING_SNAKE_CASE				= baseaa.urlsafe_baadecode(UpperCamelCase      ).decode("utf-8"      )
																				_SCREAMING_SNAKE_CASE				= json.loads(UpperCamelCase      )
														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}'      )

								_SCREAMING_SNAKE_CASE				= config

								self.set_stage_and_offload()




		def     lowercase       (				self      ):
								# 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.
								_SCREAMING_SNAKE_CASE				= self.get_value("zero_optimization.stage"					,						-1      )

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




		def     lowercase       (				self					,						UpperCamelCase      ):
								_SCREAMING_SNAKE_CASE				= self.config

								# find the config node of interest if it exists
								_SCREAMING_SNAKE_CASE				= ds_key_long.split("."      )
								_SCREAMING_SNAKE_CASE				= nodes.pop()
								for node in nodes:
														_SCREAMING_SNAKE_CASE				= config.get(UpperCamelCase      )
														if config is None:
																				return None, ds_key

								return config, ds_key




		def     lowercase       (				self					,						UpperCamelCase					,						UpperCamelCase=None      ):
								_SCREAMING_SNAKE_CASE,							_SCREAMING_SNAKE_CASE				= self.find_config_node(UpperCamelCase      )
								if config is None:
														return default
								return config.get(UpperCamelCase					,						UpperCamelCase      )




		def     lowercase       (				self					,						UpperCamelCase					,						UpperCamelCase=False      ):
								_SCREAMING_SNAKE_CASE				= self.config

								# find the config node of interest if it exists
								_SCREAMING_SNAKE_CASE				= ds_key_long.split("."      )
								for node in nodes:
														_SCREAMING_SNAKE_CASE				= config
														_SCREAMING_SNAKE_CASE				= config.get(UpperCamelCase      )
														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(UpperCamelCase      )




		def     lowercase       (				self					,						UpperCamelCase      ):
								_SCREAMING_SNAKE_CASE				= self.get_value(UpperCamelCase      )
								return False if value is None else bool(UpperCamelCase      )




		def     lowercase       (				self					,						UpperCamelCase      ):
								_SCREAMING_SNAKE_CASE				= self.get_value(UpperCamelCase      )
								return False if value is None else not bool(UpperCamelCase      )




		def     lowercase       (				self      ):
								return self._stage == 2




		def     lowercase       (				self      ):
								return self._stage == 3




		def     lowercase       (				self      ):
								return self._offload


class 			lowerCAmelCase :




		def __init__(				self					,						UpperCamelCase      ):
								_SCREAMING_SNAKE_CASE				= engine




		def     lowercase       (				self					,						UpperCamelCase					,						**UpperCamelCase      ):
								# runs backpropagation and handles mixed precision
								self.engine.backward(UpperCamelCase					,						**UpperCamelCase      )

								# 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					,						UpperCamelCase      ):
								super().__init__(UpperCamelCase					,						device_placement=UpperCamelCase					,						scaler=UpperCamelCase      )
								_SCREAMING_SNAKE_CASE				= hasattr(self.optimizer					,						"overflow"      )




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




		def     lowercase       (				self      ):
								pass  # `accelerator.backward(loss)` is doing that automatically. Therefore, its implementation is not needed




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


class 			lowerCAmelCase (				__UpperCAmelCase			):




		def __init__(				self					,						UpperCamelCase					,						UpperCamelCase      ):
								super().__init__(UpperCamelCase					,						UpperCamelCase      )




		def     lowercase       (				self      ):
								pass  # `accelerator.backward(loss)` is doing that automatically. Therefore, its implementation is not needed


class 			lowerCAmelCase :




		def __init__(				self					,						UpperCamelCase					,						UpperCamelCase=0.0_01					,						UpperCamelCase=0					,						**UpperCamelCase      ):
								_SCREAMING_SNAKE_CASE				= params
								_SCREAMING_SNAKE_CASE				= lr
								_SCREAMING_SNAKE_CASE				= weight_decay
								_SCREAMING_SNAKE_CASE				= kwargs


class 			lowerCAmelCase :




		def __init__(				self					,						UpperCamelCase					,						UpperCamelCase=None					,						UpperCamelCase=0					,						**UpperCamelCase      ):
								_SCREAMING_SNAKE_CASE				= optimizer
								_SCREAMING_SNAKE_CASE				= total_num_steps
								_SCREAMING_SNAKE_CASE				= warmup_num_steps
								_SCREAMING_SNAKE_CASE				= kwargs