






'''simple docstring'''


# Copyright 2022 The HuggingFace Team. All rights reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

import argparse
import os
import platform

import numpy as np
import psutil
import torch

from accelerate import __version__ as version
from accelerate.commands.config import default_config_file, load_config_from_file

from ..utils import is_npu_available, is_xpu_available







def 	__magic_name__( _A=None					):
						'''simple docstring'''





						if subparsers is not None:
												UpperCamelCase__         = subparsers.add_parser("""env"""					)
						else:
												UpperCamelCase__         = argparse.ArgumentParser("""Accelerate env command"""					)

						parser.add_argument(
						    """--config_file"""						,      default=_A						,      help="""The config file to use for the default values in the launching script."""					)

						if subparsers is not None:
												parser.set_defaults(func=_A					)
						return parser







def 	__magic_name__( _A					):
						'''simple docstring'''





						UpperCamelCase__         = torch.__version__
						UpperCamelCase__         = torch.cuda.is_available()
						UpperCamelCase__         = is_xpu_available()
						UpperCamelCase__         = is_npu_available()

						UpperCamelCase__         = """Not found"""
						# Get the default from the config file.
						if args.config_file is not None or os.path.isfile(_A					):
												UpperCamelCase__         = load_config_from_file(args.config_file					).to_dict()

						UpperCamelCase__         = {
						    """`Accelerate` version""": version,
						    """Platform""": platform.platform(),
						    """Python version""": platform.python_version(),
						    """Numpy version""": np.__version__,
						    """PyTorch version (GPU?)""": f"{pt_version} ({pt_cuda_available})",
						    """PyTorch XPU available""": str(_A					),
						    """PyTorch NPU available""": str(_A					),
						    """System RAM""": f"{psutil.virtual_memory().total / 1024 ** 3:.2f} GB",
						}
						if pt_cuda_available:
												UpperCamelCase__         = torch.cuda.get_device_name()

						print("""\nCopy-and-paste the text below in your GitHub issue\n"""					)
						print("""\n""".join([f"- {prop}: {val}" for prop, val in info.items()]					)					)

						print("""- `Accelerate` default config:""" if args.config_file is None else """- `Accelerate` config passed:"""					)
						UpperCamelCase__         = (
						    """\n""".join([f"\t- {prop}: {val}" for prop, val in accelerate_config.items()]					)
						    if isinstance(_A						,      _A					)
						    else f"\t{accelerate_config}"
						)
						print(_A					)

						UpperCamelCase__         = accelerate_config

						return info







def 	__magic_name__( ):
						'''simple docstring'''





						UpperCamelCase__         = env_command_parser()
						UpperCamelCase__         = parser.parse_args()
						env_command(_A					)
						return 0


if __name__ == "__main__":
					raise SystemExit(main())








'''simple docstring'''


# 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    _SCREAMING_SNAKE_CASE							:


			'''simple docstring'''





			def __init__(   self							:						Optional[int] ,      lowercase							:						Optional[int] ,      lowercase							:						Optional[Any] ,      lowercase							:						bool = True ,      lowercase							:						bool = False			)							->    Optional[int]:

									'''simple docstring'''







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






			def 				A      (   self							:						int ,      *lowercase							:						int ,      **lowercase							:						str			)							->    str:

									'''simple docstring'''







									if not self.step_with_optimizer:
															# No link between scheduler and optimizer -> just step
															self.scheduler.step(*lowercase ,      **lowercase			)
															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(*lowercase ,      **lowercase			)
									else:
															# Otherwise the training dataloader batch size was multiplied by `num_processes`, so we need to do
															# num_processes steps per training step
															UpperCamelCase__         = AcceleratorState().num_processes
															for _ in range(lowercase			):
																					# 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(*lowercase ,      **lowercase			)
																					else:
																											self.scheduler.step(*lowercase ,      **lowercase			)






			def 				A      (   self							:						List[str]			)							->    Optional[Any]:

									'''simple docstring'''







									return self.scheduler.get_last_lr()






			def 				A      (   self							:						Any			)							->    int:

									'''simple docstring'''







									return self.scheduler.state_dict()






			def 				A      (   self							:						Any ,      lowercase							:						int			)							->    Optional[int]:

									'''simple docstring'''







									self.scheduler.load_state_dict(lowercase			)






			def 				A      (   self							:						str			)							->    int:

									'''simple docstring'''







									return self.scheduler.get_lr()







			def 				A      (   self							:						Union[str, Any] ,      *lowercase							:						List[Any] ,      **lowercase							:						str			)							->    str:

									'''simple docstring'''







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