

'''simple docstring'''



import importlib
import importlib.metadata
import os
import warnings
from distutils.util import strtobool
from functools import lru_cache

import torch
from packaging import version
from packaging.version import parse

from .environment import parse_flag_from_env
from .versions import compare_versions, is_torch_version


try:
  import torch_xla.core.xla_model as xm  # noqa: F401

  UpperCAmelCase				: Tuple         =  True
except ImportError:
  UpperCAmelCase				: int         =  False


# Cache this result has it's a C FFI call which can be pretty time-consuming
UpperCAmelCase				: int         =  torch.distributed.is_available()







def      a__ (						a__		):




     """simple docstring"""


     __SCREAMING_SNAKE_CASE						     =   importlib.util.find_spec(UpperCamelCase__		) is not None
     if package_exists:
          try:
               __SCREAMING_SNAKE_CASE						     =   importlib.metadata.metadata(UpperCamelCase__		)
               return True
          except importlib.metadata.PackageNotFoundError:
               return False







def      a__ (						):




     """simple docstring"""


     return _torch_distributed_available







def      a__ (						):




     """simple docstring"""


     try:
          pass
     except ImportError:
          print(
              """Intel(R) oneCCL Bindings for PyTorch* is required to run DDP on Intel(R) GPUs, but it is not"""
              """ detected. If you see \"ValueError: Invalid backend: \'ccl\'\" error, please install Intel(R) oneCCL"""
              """ Bindings for PyTorch*."""		)
     return (
         importlib.util.find_spec("""torch_ccl"""		) is not None
         or importlib.util.find_spec("""oneccl_bindings_for_pytorch"""		) is not None
     )







def      a__ (						):




     """simple docstring"""


     return importlib.metadata.version("""oneccl_bind_pt"""		)







def      a__ (						):




     """simple docstring"""


     return _is_package_available("""transformer_engine"""		)







@lru_cache
def      a__ (						a__=True		):




     """simple docstring"""


     if torch.cuda.is_available():
          return False
     if _tpu_available and check_device:
          try:
               # Will raise a RuntimeError if no XLA configuration is found
               __SCREAMING_SNAKE_CASE						     =   xm.xla_device()
               return True
          except RuntimeError:
               return False
     return _tpu_available







def      a__ (						):




     """simple docstring"""


     return _is_package_available("""deepspeed"""		)







def      a__ (						a__=False		):




     """simple docstring"""


     if is_tpu_available():
          return not ignore_tpu
     if torch.cuda.is_available():
          return torch.cuda.is_bfaa_supported()
     if is_npu_available():
          return False
     return True







def      a__ (						):




     """simple docstring"""


     __SCREAMING_SNAKE_CASE						     =   _is_package_available("""bitsandbytes"""		)
     if package_exists:
          __SCREAMING_SNAKE_CASE						     =   version.parse(importlib.metadata.version("""bitsandbytes"""		)		)
          return compare_versions(UpperCamelCase__   ,					""">="""   ,					"""0.39.0"""		)
     return False







def      a__ (						):




     """simple docstring"""


     __SCREAMING_SNAKE_CASE						     =   _is_package_available("""bitsandbytes"""		)
     if package_exists:
          __SCREAMING_SNAKE_CASE						     =   version.parse(importlib.metadata.version("""bitsandbytes"""		)		)
          return compare_versions(UpperCamelCase__   ,					""">="""   ,					"""0.37.2"""		)
     return False







def      a__ (						):




     """simple docstring"""


     return _is_package_available("""bitsandbytes"""		)







def      a__ (						):




     """simple docstring"""


     if strtobool(os.environ.get("""ACCELERATE_USE_MEGATRON_LM"""   ,					"""False"""		)		) == 1:
          __SCREAMING_SNAKE_CASE						     =   importlib.util.find_spec("""megatron"""		) is not None
          if package_exists:
               try:
                    __SCREAMING_SNAKE_CASE						     =   parse(importlib.metadata.version("""megatron-lm"""		)		)
                    return compare_versions(UpperCamelCase__   ,					""">="""   ,					"""2.2.0"""		)
               except Exception as e:
                    warnings.warn(F'Parse Megatron version failed. Exception:{e}'		)
                    return False







def      a__ (						):




     """simple docstring"""


     return _is_package_available("""safetensors"""		)







def      a__ (						):




     """simple docstring"""


     return _is_package_available("""transformers"""		)







def      a__ (						):




     """simple docstring"""


     return _is_package_available("""datasets"""		)







def      a__ (						):




     """simple docstring"""


     return _is_package_available("""aim"""		)







def      a__ (						):




     """simple docstring"""


     return _is_package_available("""tensorboard"""		) or _is_package_available("""tensorboardX"""		)







def      a__ (						):




     """simple docstring"""


     return _is_package_available("""wandb"""		)







def      a__ (						):




     """simple docstring"""


     return _is_package_available("""comet_ml"""		)







def      a__ (						):




     """simple docstring"""


     return _is_package_available("""boto3"""		)







def      a__ (						):




     """simple docstring"""


     if _is_package_available("""rich"""		):
          if "ACCELERATE_DISABLE_RICH" in os.environ:
               warnings.warn(
                   """`ACCELERATE_DISABLE_RICH` is deprecated and will be removed in v0.22.0 and deactivated by default. Please use `ACCELERATE_ENABLE_RICH` if you wish to use `rich`."""		)
               return not parse_flag_from_env("""ACCELERATE_DISABLE_RICH"""   ,					UpperCamelCase__		)
          return parse_flag_from_env("""ACCELERATE_ENABLE_RICH"""   ,					UpperCamelCase__		)
     return False







def      a__ (						):




     """simple docstring"""


     return _is_package_available("""sagemaker"""		)







def      a__ (						):




     """simple docstring"""


     return _is_package_available("""tqdm"""		)







def      a__ (						):




     """simple docstring"""


     return _is_package_available("""mlflow"""		)







def      a__ (						):




     """simple docstring"""


     return is_torch_version(""">="""   ,					"""1.12"""		) and torch.backends.mps.is_available() and torch.backends.mps.is_built()







def      a__ (						):




     """simple docstring"""


     def get_major_and_minor_from_version(a__		):
          return str(version.parse(UpperCamelCase__		).major		) + "." + str(version.parse(UpperCamelCase__		).minor		)

     __SCREAMING_SNAKE_CASE						     =   importlib.metadata.version("""torch"""		)
     if importlib.util.find_spec("""intel_extension_for_pytorch"""		) is None:
          return False
     __SCREAMING_SNAKE_CASE						     =   '''N/A'''
     try:
          __SCREAMING_SNAKE_CASE						     =   importlib.metadata.version("""intel_extension_for_pytorch"""		)
     except importlib.metadata.PackageNotFoundError:
          return False
     __SCREAMING_SNAKE_CASE						     =   get_major_and_minor_from_version(_torch_version		)
     __SCREAMING_SNAKE_CASE						     =   get_major_and_minor_from_version(_ipex_version		)
     if torch_major_and_minor != ipex_major_and_minor:
          warnings.warn(
              F'Intel Extension for PyTorch {ipex_major_and_minor} needs to work with PyTorch {ipex_major_and_minor}.*,'
              F' but PyTorch {_torch_version} is found. Please switch to the matching version and run again.'		)
          return False
     return True







@lru_cache
def      a__ (						a__=False		):




     """simple docstring"""


     if importlib.util.find_spec("""torch"""		) is None or importlib.util.find_spec("""torch_npu"""		) is None:
          return False

     import torch
     import torch_npu  # noqa: F401

     if check_device:
          try:
               # Will raise a RuntimeError if no NPU is found
               __SCREAMING_SNAKE_CASE						     =   torch.npu.device_count()
               return torch.npu.is_available()
          except RuntimeError:
               return False
     return hasattr(UpperCamelCase__   ,					"""npu"""		) and torch.npu.is_available()







@lru_cache
def      a__ (						a__=False		):




     """simple docstring"""


     if not parse_flag_from_env("""ACCELERATE_USE_XPU"""   ,					default=UpperCamelCase__		):
          return False
     if is_ipex_available():
          import torch

          if is_torch_version("""<="""   ,					"""1.12"""		):
               return False
     else:
          return False

     import intel_extension_for_pytorch  # noqa: F401

     if check_device:
          try:
               # Will raise a RuntimeError if no XPU  is found
               __SCREAMING_SNAKE_CASE						     =   torch.xpu.device_count()
               return torch.xpu.is_available()
          except RuntimeError:
               return False
     return hasattr(UpperCamelCase__   ,					"""xpu"""		) and torch.xpu.is_available()








'''simple docstring'''



import inspect
import warnings

import torch

from .state import AcceleratorState, GradientState
from .utils import DistributedType, honor_type, is_tpu_available


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







def      a__ (						a__   ,					a__		):




     """simple docstring"""


     if isinstance(a__   ,					(list, tuple)		):
          return honor_type(a__   ,					(move_to_device(a__   ,					a__		) for t in state)		)
     elif isinstance(a__   ,					a__		):
          return type(a__		)({k: move_to_device(a__   ,					a__		) for k, v in state.items()}		)
     elif isinstance(a__   ,					torch.Tensor		):
          return state.to(a__		)
     return state







class 		lowerCAmelCase__     (   torch.optim.Optimizer      ):



 """simple docstring"""



 def __init__(	self      : Tuple      ,							__SCREAMING_SNAKE_CASE      : Union[str, Any]      ,							__SCREAMING_SNAKE_CASE      : Dict=True      ,							__SCREAMING_SNAKE_CASE      : List[Any]=None    )						->				str:


      """simple docstring"""




      __SCREAMING_SNAKE_CASE						     =   optimizer
      __SCREAMING_SNAKE_CASE						     =   scaler
      __SCREAMING_SNAKE_CASE						     =   AcceleratorState()
      __SCREAMING_SNAKE_CASE						     =   GradientState()
      __SCREAMING_SNAKE_CASE						     =   device_placement
      __SCREAMING_SNAKE_CASE						     =   False
      __SCREAMING_SNAKE_CASE						     =   None

      # Handle device placement
      if device_placement:
           __SCREAMING_SNAKE_CASE						     =   self.optimizer.state_dict()
           if self.accelerator_state.distributed_type == DistributedType.TPU:
                xm.send_cpu_data_to_device(__SCREAMING_SNAKE_CASE      ,							self.accelerator_state.device    )
           else:
                __SCREAMING_SNAKE_CASE						     =   move_to_device(__SCREAMING_SNAKE_CASE      ,							self.accelerator_state.device    )
           self.optimizer.load_state_dict(__SCREAMING_SNAKE_CASE    )
 @property
 def    UpperCAmelCase__	(	self      : str    )						->				Union[str, Any]:


      """simple docstring"""




      return self.optimizer.state
 @state.setter
 def    UpperCAmelCase__	(	self      : List[str]      ,							__SCREAMING_SNAKE_CASE      : Optional[Any]    )						->				List[Any]:


      """simple docstring"""




      __SCREAMING_SNAKE_CASE						     =   state
 @property
 def    UpperCAmelCase__	(	self      : Optional[int]    )						->				str:


      """simple docstring"""




      return self.optimizer.param_groups
 @param_groups.setter
 def    UpperCAmelCase__	(	self      : Dict      ,							__SCREAMING_SNAKE_CASE      : str    )						->				List[Any]:


      """simple docstring"""




      __SCREAMING_SNAKE_CASE						     =   param_groups
 @property
 def    UpperCAmelCase__	(	self      : List[Any]    )						->				Any:


      """simple docstring"""




      return self.optimizer.defaults
 @defaults.setter
 def    UpperCAmelCase__	(	self      : Any      ,							__SCREAMING_SNAKE_CASE      : Dict    )						->				Optional[Any]:


      """simple docstring"""




      __SCREAMING_SNAKE_CASE						     =   defaults
 def    UpperCAmelCase__	(	self      : Any      ,							__SCREAMING_SNAKE_CASE      : Any    )						->				List[str]:


      """simple docstring"""




      self.optimizer.add_param_group(__SCREAMING_SNAKE_CASE    )
 def    UpperCAmelCase__	(	self      : List[str]      ,							__SCREAMING_SNAKE_CASE      : List[str]    )						->				Dict:


      """simple docstring"""




      if self.accelerator_state.distributed_type == DistributedType.TPU and self.device_placement:
           xm.send_cpu_data_to_device(__SCREAMING_SNAKE_CASE      ,							self.accelerator_state.device    )
      self.optimizer.load_state_dict(__SCREAMING_SNAKE_CASE    )
 def    UpperCAmelCase__	(	self      : Union[str, Any]    )						->				Union[str, Any]:


      """simple docstring"""




      return self.optimizer.state_dict()
 def    UpperCAmelCase__	(	self      : List[str]      ,							__SCREAMING_SNAKE_CASE      : List[str]=None    )						->				Any:


      """simple docstring"""




      if self.gradient_state.sync_gradients:
           __SCREAMING_SNAKE_CASE						     =   """set_to_none""" in inspect.signature(self.optimizer.zero_grad    ).parameters
           if accept_arg:
                if set_to_none is None:
                     __SCREAMING_SNAKE_CASE						     =   False
                self.optimizer.zero_grad(set_to_none=__SCREAMING_SNAKE_CASE    )
           else:
                if set_to_none is not None:
                     raise ValueError("""`set_to_none` for Optimizer.zero_grad` is not supported by this optimizer."""    )
                self.optimizer.zero_grad()
 def    UpperCAmelCase__	(	self      : List[str]      ,							__SCREAMING_SNAKE_CASE      : List[Any]=None    )						->				Dict:


      """simple docstring"""




      if self.gradient_state.sync_gradients:
           if self.accelerator_state.distributed_type == DistributedType.TPU:
                __SCREAMING_SNAKE_CASE						     =   {"""closure""": closure} if closure is not None else {}
                xm.optimizer_step(self.optimizer      ,							optimizer_args=__SCREAMING_SNAKE_CASE    )
           elif self.scaler is not None:
                __SCREAMING_SNAKE_CASE						     =   False
                if self._last_scale is None:
                     # `get_scale` is an async operation requiring full synchronization
                     # on CPU and GPUs before finishing. As a result, we store away
                     # the prior one to reduce the call overhead
                     __SCREAMING_SNAKE_CASE						     =   self.scaler.get_scale()
                     __SCREAMING_SNAKE_CASE						     =   True
                self.scaler.step(self.optimizer      ,							__SCREAMING_SNAKE_CASE    )
                self.scaler.update()
                __SCREAMING_SNAKE_CASE						     =   self.scaler.get_scale()
                if not new_scale:
                     # If we reduced the loss scale, it means the optimizer step was skipped because of gradient overflow.
                     __SCREAMING_SNAKE_CASE						     =   scale_after < self._last_scale
                __SCREAMING_SNAKE_CASE						     =   scale_after

           else:
                self.optimizer.step(__SCREAMING_SNAKE_CASE    )
 def    UpperCAmelCase__	(	self      : Union[str, Any]      ,							__SCREAMING_SNAKE_CASE      : Any    )						->				Any:


      """simple docstring"""




      for param_group in self.optimizer.param_groups:
           __SCREAMING_SNAKE_CASE						     =   [parameters_map.get(__SCREAMING_SNAKE_CASE      ,							__SCREAMING_SNAKE_CASE    ) for p in param_group["""params"""]]
 @property
 def    UpperCAmelCase__	(	self      : int    )						->				Any:


      """simple docstring"""




      warnings.warn(
          """The `is_overflow` property is deprecated and will be removed in version 1.0 of Accelerate use """
          """`optimizer.step_was_skipped` instead."""      ,							__SCREAMING_SNAKE_CASE      ,							)
      return self._is_overflow
 @property
 def    UpperCAmelCase__	(	self      : Union[str, Any]    )						->				Optional[int]:


      """simple docstring"""




      return self._is_overflow
 def __getstate__(	self      : Tuple    )						->				str:


      """simple docstring"""




      return self.__dict__.copy()


 def __setstate__(	self      : str      ,							__SCREAMING_SNAKE_CASE      : Optional[int]    )						->				Any:


      """simple docstring"""




      self.__dict__.update(__SCREAMING_SNAKE_CASE    )






