






# We ignore warnings about stepping the scheduler since we step it ourselves during gradient accumulation

import warnings

from .state import AcceleratorState, GradientState


warnings.filterwarnings("""ignore""", category=UserWarning, module="""torch.optim.lr_scheduler""")


class    _lowerCAmelCase			:



				"""simple docstring"""



				def __init__(      self    :    Optional[Any]			,		UpperCamelCase__    :    str			,		UpperCamelCase__    :    List[str]			,		UpperCamelCase__    :    bool = True			,		UpperCamelCase__    :    bool = False):






											'''simple docstring'''







											snake_case__	  =       scheduler
											snake_case__	  =       optimizers if isinstance(__lowerCamelCase			,		(list, tuple)) else [optimizers]
											snake_case__	  =       split_batches
											snake_case__	  =       step_with_optimizer
											snake_case__	  =       GradientState()



				def 					__magic_name__							(      self    :    Union[str, Any]			,		*UpperCamelCase__    :    int			,		**UpperCamelCase__    :    str):






											'''simple docstring'''







											if not self.step_with_optimizer:
																		# No link between scheduler and optimizer -> just step
																		self.scheduler.step(*__lowerCamelCase			,		**__lowerCamelCase)
																		return

											# Otherwise, first make sure the optimizer was stepped.
											if not self.gradient_state.sync_gradients:
																		if self.gradient_state.adjust_scheduler:
																									self.scheduler._step_count += 1
																		return

											for opt in self.optimizers:
																		if opt.step_was_skipped:
																									return
											if self.split_batches:
																		# Split batches -> the training dataloader batch size is not changed so one step per training step
																		self.scheduler.step(*__lowerCamelCase			,		**__lowerCamelCase)
											else:
																		# Otherwise the training dataloader batch size was multiplied by `num_processes`, so we need to do
																		# num_processes steps per training step
																		snake_case__	  =       AcceleratorState().num_processes
																		for _ in range(__lowerCamelCase):
																									# Special case when using OneCycle and `drop_last` was not used
																									if hasattr(self.scheduler			,		"""total_steps"""):
																																if self.scheduler._step_count <= self.scheduler.total_steps:
																																							self.scheduler.step(*__lowerCamelCase			,		**__lowerCamelCase)
																									else:
																																self.scheduler.step(*__lowerCamelCase			,		**__lowerCamelCase)



				def 					__magic_name__							(      self    :    List[str]):






											'''simple docstring'''







											return self.scheduler.get_last_lr()



				def 					__magic_name__							(      self    :    Any):






											'''simple docstring'''







											return self.scheduler.state_dict()



				def 					__magic_name__							(      self    :    Tuple			,		UpperCamelCase__    :    Optional[Any]):






											'''simple docstring'''







											self.scheduler.load_state_dict(__lowerCamelCase)



				def 					__magic_name__							(      self    :    List[Any]):






											'''simple docstring'''







											return self.scheduler.get_lr()



				def 					__magic_name__							(      self    :    Union[str, Any]			,		*UpperCamelCase__    :    str			,		**UpperCamelCase__    :    str):






											'''simple docstring'''







											return self.scheduler.print_lr(*__lowerCamelCase			,		**__lowerCamelCase)








from dataclasses import dataclass, field
from typing import Tuple

from ..utils import cached_property, is_torch_available, is_torch_tpu_available, logging, requires_backends
from .benchmark_args_utils import BenchmarkArguments


if is_torch_available():
	import torch

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


a__       =     logging.get_logger(__name__)


@dataclass
class    _lowerCAmelCase			(		lowercase_      ):



				"""simple docstring"""



				_lowercase				:   Dict      =  [
				    '''no_inference''',
				    '''no_cuda''',
				    '''no_tpu''',
				    '''no_speed''',
				    '''no_memory''',
				    '''no_env_print''',
				    '''no_multi_process''',
				]



				def __init__(      self    :    str			,		**UpperCamelCase__    :    Dict):






											'''simple docstring'''







											for deprecated_arg in self.deprecated_args:
																		if deprecated_arg in kwargs:
																									snake_case__	  =       deprecated_arg[3:]
																									setattr(self			,		UpperCamelCase__			,		not kwargs.pop(UpperCamelCase__))
																									logger.warning(
																									    F'''{deprecated_arg} is depreciated. Please use --no_{positive_arg} or'''
																									    F''' {positive_arg}={kwargs[positive_arg]}''')

											snake_case__	  =       kwargs.pop("""torchscript"""			,		self.torchscript)
											snake_case__	  =       kwargs.pop("""torch_xla_tpu_print_metrics"""			,		self.torch_xla_tpu_print_metrics)
											snake_case__	  =       kwargs.pop("""fp16_opt_level"""			,		self.fpaa_opt_level)
											super().__init__(**UpperCamelCase__)

				_lowercase				:   bool      =  field(default=lowercase_     ,      metadata={'''help''': '''Trace the models using torchscript'''}      )
				_lowercase				:   bool      =  field(default=lowercase_     ,      metadata={'''help''': '''Print Xla/PyTorch tpu metrics'''}      )
				_lowercase				:   str      =  field(
				    default='''O1'''     ,      metadata={
				        '''help''': (
				            '''For fp16: Apex AMP optimization level selected in [\'O0\', \'O1\', \'O2\', and \'O3\']. '''
				            '''See details at https://nvidia.github.io/apex/amp.html'''
				        )
				    }     ,      )



				@cached_property
				def 					__magic_name__							(      self    :    Tuple):






											'''simple docstring'''







											requires_backends(self			,		["""torch"""])
											logger.info("""PyTorch: setting up devices""")
											if not self.cuda:
																		snake_case__	  =       torch.device("""cpu""")
																		snake_case__	  =       0
											elif is_torch_tpu_available():
																		snake_case__	  =       xm.xla_device()
																		snake_case__	  =       0
											else:
																		snake_case__	  =       torch.device("""cuda""" if torch.cuda.is_available() else """cpu""")
																		snake_case__	  =       torch.cuda.device_count()
											return device, n_gpu



				@property
				def 					__magic_name__							(      self    :    Union[str, Any]):






											'''simple docstring'''







											return is_torch_tpu_available() and self.tpu



				@property
				def 					__magic_name__							(      self    :    List[str]):






											'''simple docstring'''







											requires_backends(self			,		["""torch"""])
											# TODO(PVP): currently only single GPU is supported
											return torch.cuda.current_device()



				@property
				def 					__magic_name__							(      self    :    Union[str, Any]):






											'''simple docstring'''







											requires_backends(self			,		["""torch"""])
											return self._setup_devices[0]



				@property
				def 					__magic_name__							(      self    :    str):






											'''simple docstring'''







											requires_backends(self			,		["""torch"""])
											return self._setup_devices[1]



				@property
				def 					__magic_name__							(      self    :    str):






											'''simple docstring'''







											return self.n_gpu > 0
