


"""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

		a_											=     True
except ImportError:
		a_											=     False


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






def   a__			(					__lowercase )						->						Tuple:
				_A       	=      importlib.util.find_spec(_a ) is not None
				if package_exists:
								try:
												_A       	=      importlib.metadata.metadata(_a )
												return True
								except importlib.metadata.PackageNotFoundError:
												return False






def   a__			(					)						->						Dict:
				return _torch_distributed_available






def   a__			(					)						->						List[str]:




				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__			(					)						->						Optional[Any]:
				return importlib.metadata.version("oneccl_bind_pt" )






def   a__			(					)						->						Any:
				return _is_package_available("transformer_engine" )






@lru_cache
def   a__			(					__lowercase=True )						->						Union[str, Any]:




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






def   a__			(					)						->						Tuple:
				return _is_package_available("deepspeed" )






def   a__			(					__lowercase=False )						->						List[str]:




				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__			(					)						->						Optional[Any]:
				_A       	=      _is_package_available("bitsandbytes" )
				if package_exists:
								_A       	=      version.parse(importlib.metadata.version("bitsandbytes" ) )
								return compare_versions(_a  ,      ">="  ,      "0.39.0" )
				return False






def   a__			(					)						->						Tuple:
				_A       	=      _is_package_available("bitsandbytes" )
				if package_exists:
								_A       	=      version.parse(importlib.metadata.version("bitsandbytes" ) )
								return compare_versions(_a  ,      ">="  ,      "0.37.2" )
				return False






def   a__			(					)						->						str:
				return _is_package_available("bitsandbytes" )






def   a__			(					)						->						str:




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






def   a__			(					)						->						str:
				return _is_package_available("safetensors" )






def   a__			(					)						->						Tuple:
				return _is_package_available("transformers" )






def   a__			(					)						->						Any:
				return _is_package_available("datasets" )






def   a__			(					)						->						Any:
				return _is_package_available("aim" )






def   a__			(					)						->						Tuple:
				return _is_package_available("tensorboard" ) or _is_package_available("tensorboardX" )






def   a__			(					)						->						int:
				return _is_package_available("wandb" )






def   a__			(					)						->						str:
				return _is_package_available("comet_ml" )






def   a__			(					)						->						List[Any]:
				return _is_package_available("boto3" )






def   a__			(					)						->						Optional[int]:




				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"  ,      _a )
								return parse_flag_from_env("ACCELERATE_ENABLE_RICH"  ,      _a )
				return False






def   a__			(					)						->						Optional[Any]:
				return _is_package_available("sagemaker" )






def   a__			(					)						->						Union[str, Any]:
				return _is_package_available("tqdm" )






def   a__			(					)						->						Optional[Any]:
				return _is_package_available("mlflow" )






def   a__			(					)						->						Union[str, Any]:
				return is_torch_version(">="  ,      "1.12" ) and torch.backends.mps.is_available() and torch.backends.mps.is_built()






def   a__			(					)						->						Optional[int]:




				def get_major_and_minor_from_version(__lowercase ):
								return str(version.parse(_a ).major ) + "." + str(version.parse(_a ).minor )

				_A       	=      importlib.metadata.version("torch" )
				if importlib.util.find_spec("intel_extension_for_pytorch" ) is None:
								return False
				_A       	=      "N/A"
				try:
								_A       	=      importlib.metadata.version("intel_extension_for_pytorch" )
				except importlib.metadata.PackageNotFoundError:
								return False
				_A       	=      get_major_and_minor_from_version(_torch_version )
				_A       	=      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__			(					__lowercase=False )						->						int:




				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
												_A       	=      torch.npu.device_count()
												return torch.npu.is_available()
								except RuntimeError:
												return False
				return hasattr(_a  ,      "npu" ) and torch.npu.is_available()






@lru_cache
def   a__			(					__lowercase=False )						->						Tuple:




				if not parse_flag_from_env("ACCELERATE_USE_XPU"  ,      default=_a ):
								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
												_A       	=      torch.xpu.device_count()
												return torch.xpu.is_available()
								except RuntimeError:
												return False
				return hasattr(_a  ,      "xpu" ) and torch.xpu.is_available()

class 		__magic_name__				:


  """simple docstring"""





  class 		__magic_name__				:


    """simple docstring"""





    _UpperCamelCase						 = "_prev", "_data", "_next"


    def __init__(					self	,		a__	,		a__	,		a__		):
      _lowerCamelCase      =						link_p
      _lowerCamelCase      =						element
      _lowerCamelCase      =						link_n


    def     _UpperCAmelCase  (					self		):
      return (
          f''' Prev -> {self._prev is not None}, Next -> {self._next is not None}'''
      )







  def __init__(					self		):
    _lowerCamelCase      =						self._Node(a__	,		a__	,		a__		)
    _lowerCamelCase      =						self._Node(a__	,		a__	,		a__		)
    _lowerCamelCase      =						self._trailer
    _lowerCamelCase      =						self._header
    _lowerCamelCase      =						0


  def __len__(					self		):
    return self._size


  def     _UpperCAmelCase  (					self		):
    return self.__len__() == 0


  def     _UpperCAmelCase  (					self	,		a__	,		a__	,		a__		):
    # Create new_node by setting it's prev.link -> header
    # setting it's next.link -> trailer
    _lowerCamelCase      =						self._Node(a__	,		a__	,		a__		)
    _lowerCamelCase      =						new_node
    _lowerCamelCase      =						new_node
    self._size += 1
    return self


  def     _UpperCAmelCase  (					self	,		a__		):
    _lowerCamelCase      =						node._prev
    _lowerCamelCase      =						node._next

    _lowerCamelCase      =						successor
    _lowerCamelCase      =						predecessor
    self._size -= 1
    _lowerCamelCase      =						node._data
    _lowerCamelCase      =						_lowerCamelCase      =						_lowerCamelCase      =						None
    del node
    return temp







class 		__magic_name__				( _DoublyLinkedBase					):


  """simple docstring"""





  def     _UpperCAmelCase  (					self		):
    if self.is_empty():
      raise Exception('''List is empty'''		)
    return self._header._next._data


  def     _UpperCAmelCase  (					self		):
    if self.is_empty():
      raise Exception('''List is empty'''		)
    return self._trailer._prev._data


  def     _UpperCAmelCase  (					self	,		a__		):
    return self._insert(self._header	,		a__	,		self._header._next		)


  def     _UpperCAmelCase  (					self	,		a__		):
    return self._insert(self._trailer._prev	,		a__	,		self._trailer		)


  def     _UpperCAmelCase  (					self		):
    if self.is_empty():
      raise IndexError('''remove_first from empty list'''		)
    return self._delete(self._header._next		)


  def     _UpperCAmelCase  (					self		):
    if self.is_empty():
      raise IndexError('''remove_first from empty list'''		)
    return self._delete(self._trailer._prev		)




