



"""simple docstring"""

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

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







class    __lowercase     :







						"""simple docstring"""







						def __init__(self   ,		lowercase__       ):
											if isinstance(lowercase__   ,		lowercase__       ):
																# 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_					:		Optional[Any] 					=					deepcopy(lowercase__       )
											elif os.path.exists(lowercase__       ):
																with io.open(lowercase__   ,		"""r"""   ,		encoding="""utf-8"""       ) as f:
																					snake_case_					:		Optional[Any] 					=					json.load(lowercase__       )
											else:
																try:
																					snake_case_					:		Optional[int] 					=					baseaa.urlsafe_baadecode(lowercase__       ).decode("""utf-8"""       )
																					snake_case_					:		Tuple 					=					json.loads(lowercase__       )
																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_					:		int 					=					config

											self.set_stage_and_offload()







						def 						__UpperCamelCase							(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.
											snake_case_					:		Optional[Any] 					=					self.get_value("""zero_optimization.stage"""   ,		-1       )

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







						def 						__UpperCamelCase							(self   ,		lowercase__       ):
											snake_case_					:		Dict 					=					self.config

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

											return config, ds_key







						def 						__UpperCamelCase							(self   ,		lowercase__   ,		lowercase__=None       ):
											snake_case_					,			snake_case_					:		Dict 					=					self.find_config_node(lowercase__       )
											if config is None:
																return default
											return config.get(lowercase__   ,		lowercase__       )







						def 						__UpperCamelCase							(self   ,		lowercase__   ,		lowercase__=False       ):
											snake_case_					:		Any 					=					self.config

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







						def 						__UpperCamelCase							(self   ,		lowercase__       ):
											snake_case_					:		Any 					=					self.get_value(lowercase__       )
											return False if value is None else bool(lowercase__       )







						def 						__UpperCamelCase							(self   ,		lowercase__       ):
											snake_case_					:		Dict 					=					self.get_value(lowercase__       )
											return False if value is None else not bool(lowercase__       )







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







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







						def 						__UpperCamelCase							(self       ):
											return self._offload







class    __lowercase     :







						"""simple docstring"""







						def __init__(self   ,		lowercase__       ):
											snake_case_					:		Tuple 					=					engine







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

											# 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    __lowercase     (					_UpperCAmelCase):







						"""simple docstring"""







						def __init__(self   ,		lowercase__       ):
											super().__init__(lowercase__   ,		device_placement=lowercase__   ,		scaler=lowercase__       )
											snake_case_					:		Tuple 					=					hasattr(self.optimizer   ,		"""overflow"""       )







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







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







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







class    __lowercase     (					_UpperCAmelCase):







						"""simple docstring"""







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







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







class    __lowercase     :







						"""simple docstring"""







						def __init__(self   ,		lowercase__   ,		lowercase__=0.001   ,		lowercase__=0   ,		**lowercase__       ):
											snake_case_					:		Optional[int] 					=					params
											snake_case_					:		Optional[int] 					=					lr
											snake_case_					:		Dict 					=					weight_decay
											snake_case_					:		Optional[int] 					=					kwargs







class    __lowercase     :







						"""simple docstring"""







						def __init__(self   ,		lowercase__   ,		lowercase__=None   ,		lowercase__=0   ,		**lowercase__       ):
											snake_case_					:		Dict 					=					optimizer
											snake_case_					:		Optional[Any] 					=					total_num_steps
											snake_case_					:		List[str] 					=					warmup_num_steps
											snake_case_					:		Any 					=					kwargs









"""simple docstring"""

import copy
from dataclasses import dataclass, field
from typing import ClassVar, Dict

from ..features import Audio, Features, Value
from .base import TaskTemplate







@dataclass(frozen=_UpperCAmelCase)
class    __lowercase     (					_UpperCAmelCase):







						"""simple docstring"""







						_A		:							str							       =		field(default="""automatic-speech-recognition"""					,    metadata={"""include_in_asdict_even_if_is_default""": True})
						_A		:							ClassVar[Features]							       =		Features({"""audio""": Audio()})
						_A		:							ClassVar[Features]							       =		Features({"""transcription""": Value("""string""")})
						_A		:							str							       =		"audio"
						_A		:							str							       =		"transcription"







						def 						__UpperCamelCase							(self   ,		lowercase__       ):
											if self.audio_column not in features:
																raise ValueError(f'Column {self.audio_column} is not present in features.'       )
											if not isinstance(features[self.audio_column]   ,		lowercase__       ):
																raise ValueError(f'Column {self.audio_column} is not an Audio type.'       )
											snake_case_					:		Optional[int] 					=					copy.deepcopy(self       )
											snake_case_					:		Tuple 					=					self.input_schema.copy()
											snake_case_					:		List[str] 					=					features[self.audio_column]
											snake_case_					:		Any 					=					input_schema
											return task_template







						@property
						def 						__UpperCamelCase							(self       ):
											return {self.audio_column: "audio", self.transcription_column: "transcription"}




