



# Copyright 2021 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 json
import os
from dataclasses import dataclass
from enum import Enum
from typing import List, Optional, Union

import yaml

from ...utils import ComputeEnvironment, DistributedType, SageMakerDistributedType
from ...utils.constants import SAGEMAKER_PYTHON_VERSION, SAGEMAKER_PYTORCH_VERSION, SAGEMAKER_TRANSFORMERS_VERSION


A_		:  Dict										=  os.path.expanduser(
    os.getenv('''HF_HOME''', os.path.join(os.getenv('''XDG_CACHE_HOME''', '''~/.cache'''), '''huggingface'''))
)
A_		:  List[Any]										=  os.path.join(hf_cache_home, '''accelerate''')
A_		:  Any										=  os.path.join(cache_dir, '''default_config.yaml''')
A_		:  Optional[Any]										=  os.path.join(cache_dir, '''default_config.yaml''')

# For backward compatibility: the default config is the json one if it's the only existing file.
if os.path.isfile(default_yaml_config_file) or not os.path.isfile(default_json_config_file):
     A_		:  str										=  default_yaml_config_file
else:
     A_		:  List[str]										=  default_json_config_file


def 						UpperCAmelCase__   (   UpperCAmelCase__			:Dict       ):






 '''simple docstring'''

 if config_file is not None:
  if not os.path.isfile(UpperCAmelCase__       ):
   raise FileNotFoundError(
       F"""The passed configuration file `{config_file}` does not exist. """
       "Please pass an existing file to `accelerate launch`, or use the the default one "
       "created through `accelerate config` and run `accelerate launch` "
       "without the `--config_file` argument."       )
 else:
  a							=		default_config_file
 with open(UpperCAmelCase__			,       "r"			,       encoding="utf-8"       ) as f:
  if config_file.endswith(".json"       ):
   if (
       json.load(UpperCAmelCase__       ).get("compute_environment"			,       ComputeEnvironment.LOCAL_MACHINE       )
       == ComputeEnvironment.LOCAL_MACHINE
   ):
    a							=		ClusterConfig
   else:
    a							=		SageMakerConfig
   return config_class.from_json_file(json_file=UpperCAmelCase__       )
  else:
   if (
       yaml.safe_load(UpperCAmelCase__       ).get("compute_environment"			,       ComputeEnvironment.LOCAL_MACHINE       )
       == ComputeEnvironment.LOCAL_MACHINE
   ):
    a							=		ClusterConfig
   else:
    a							=		SageMakerConfig
   return config_class.from_yaml_file(yaml_file=UpperCAmelCase__       )



@dataclass
class 		_lowercase		:
   _UpperCAmelCase  						=      42
   _UpperCAmelCase  						=      42
   _UpperCAmelCase  						=      42
   _UpperCAmelCase  						=      42

   def 		A				(   self      :							int				)							->					str:


    """simple docstring"""






    a							=		self.__dict__
    # For serialization, it's best to convert Enums to strings (or their underlying value type).
    for key, value in result.items():
     if isinstance(__lowerCAmelCase				, __lowerCAmelCase				):
      a							=		value.value
     if isinstance(__lowerCAmelCase				, __lowerCAmelCase				) and not bool(__lowerCAmelCase				):
      a							=		None
    a							=		{k: v for k, v in result.items() if v is not None}
    return result

   @classmethod
   def 		A				(   cls      :							Optional[int]				, __lowerCAmelCase      :							Tuple=None				)							->					Union[str, Any]:


    """simple docstring"""






    a							=		default_json_config_file if json_file is None else json_file
    with open(__lowerCAmelCase				, "r"				, encoding="utf-8"				) as f:
     a							=		json.load(__lowerCAmelCase				)
    if "compute_environment" not in config_dict:
     a							=		ComputeEnvironment.LOCAL_MACHINE
    if "mixed_precision" not in config_dict:
     a							=		"fp16" if ("fp16" in config_dict and config_dict["fp16"]) else None
    if "fp16" in config_dict:  # Convert the config to the new format.
     del config_dict["fp16"]
    if "dynamo_backend" in config_dict:  # Convert the config to the new format.
     a							=		config_dict.pop("dynamo_backend"				)
     a							=		{} if dynamo_backend == "NO" else {"dynamo_backend": dynamo_backend}
    if "use_cpu" not in config_dict:
     a							=		False
    return cls(**__lowerCAmelCase				)

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


    """simple docstring"""






    with open(__lowerCAmelCase				, "w"				, encoding="utf-8"				) as f:
     a							=		json.dumps(self.to_dict()				, indent=2				, sort_keys=__lowerCAmelCase				) + "\n"
     f.write(__lowerCAmelCase				)

   @classmethod
   def 		A				(   cls      :							Optional[int]				, __lowerCAmelCase      :							List[str]=None				)							->					Optional[int]:


    """simple docstring"""






    a							=		default_yaml_config_file if yaml_file is None else yaml_file
    with open(__lowerCAmelCase				, "r"				, encoding="utf-8"				) as f:
     a							=		yaml.safe_load(__lowerCAmelCase				)
    if "compute_environment" not in config_dict:
     a							=		ComputeEnvironment.LOCAL_MACHINE

    if "mixed_precision" not in config_dict:
     a							=		"fp16" if ("fp16" in config_dict and config_dict["fp16"]) else None
    if isinstance(config_dict["mixed_precision"]				, __lowerCAmelCase				) and not config_dict["mixed_precision"]:
     a							=		"no"
    if "fp16" in config_dict:  # Convert the config to the new format.
     del config_dict["fp16"]
    if "dynamo_backend" in config_dict:  # Convert the config to the new format.
     a							=		config_dict.pop("dynamo_backend"				)
     a							=		{} if dynamo_backend == "NO" else {"dynamo_backend": dynamo_backend}
    if "use_cpu" not in config_dict:
     a							=		False
    return cls(**__lowerCAmelCase				)

   def 		A				(   self      :							int				, __lowerCAmelCase      :							List[str]				)							->					Optional[int]:


    """simple docstring"""






    with open(__lowerCAmelCase				, "w"				, encoding="utf-8"				) as f:
     yaml.safe_dump(self.to_dict()				, __lowerCAmelCase				)




   def 		A				(   self      :							int				)							->					int:


    """simple docstring"""






    if isinstance(self.compute_environment				, __lowerCAmelCase				):
     a							=		ComputeEnvironment(self.compute_environment				)
    if isinstance(self.distributed_type				, __lowerCAmelCase				):
     if self.compute_environment == ComputeEnvironment.AMAZON_SAGEMAKER:
      a							=		SageMakerDistributedType(self.distributed_type				)
     else:
      a							=		DistributedType(self.distributed_type				)
    if self.dynamo_config is None:
     a							=		{}



@dataclass
class 		_lowercase		(							UpperCAmelCase__						):
   _UpperCAmelCase  						=      42
   _UpperCAmelCase  						=      0
   _UpperCAmelCase  						=      1
   _UpperCAmelCase  						=      None
   _UpperCAmelCase  						=      None
   _UpperCAmelCase  						=      None
   _UpperCAmelCase  						=      "static"
   _UpperCAmelCase  						=      False
   _UpperCAmelCase  						=      "main"

   # args for deepspeed_plugin
   _UpperCAmelCase  						=      None
   # args for fsdp
   _UpperCAmelCase  						=      None
   # args for megatron_lm
   _UpperCAmelCase  						=      None
   # args for ipex
   _UpperCAmelCase  						=      None
   # args for TPU
   _UpperCAmelCase  						=      False

   # args for TPU pods
   _UpperCAmelCase  						=      None
   _UpperCAmelCase  						=      None
   _UpperCAmelCase  						=      False
   _UpperCAmelCase  						=      False
   _UpperCAmelCase  						=      None
   _UpperCAmelCase  						=      None
   _UpperCAmelCase  						=      None
   _UpperCAmelCase  						=      None

   # args for dynamo
   _UpperCAmelCase  						=      None

   def 		A				(   self      :							Dict				)							->					Optional[int]:


    """simple docstring"""






    if self.deepspeed_config is None:
     a							=		{}
    if self.fsdp_config is None:
     a							=		{}
    if self.megatron_lm_config is None:
     a							=		{}
    if self.ipex_config is None:
     a							=		{}
    return super().__post_init__()



@dataclass
class 		_lowercase		(							UpperCAmelCase__						):
   _UpperCAmelCase  						=      42
   _UpperCAmelCase  						=      42
   _UpperCAmelCase  						=      None
   _UpperCAmelCase  						=      None
   _UpperCAmelCase  						=      "us-east-1"
   _UpperCAmelCase  						=      1
   _UpperCAmelCase  						=      "all"
   _UpperCAmelCase  						=      F"""accelerate-sagemaker-{num_machines}"""
   _UpperCAmelCase  						=      SAGEMAKER_PYTORCH_VERSION
   _UpperCAmelCase  						=      SAGEMAKER_TRANSFORMERS_VERSION
   _UpperCAmelCase  						=      SAGEMAKER_PYTHON_VERSION
   _UpperCAmelCase  						=      None
   _UpperCAmelCase  						=      None
   _UpperCAmelCase  						=      None
   _UpperCAmelCase  						=      None







'''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			     =					True
except ImportError:
     UpperCamelCase			     =					False


# Cache this result has it's a C FFI call which can be pretty time-consuming
UpperCamelCase			     =					torch.distributed.is_available()
def 					SCREAMING_SNAKE_CASE(   __lowercase	)			-> Optional[int]:
     # Check we're not importing a "pkg_name" directory somewhere but the actual library by trying to grab the version
     A:		Optional[Any]          =		importlib.util.find_spec(__lowercase	) is not None
     if package_exists:
          try:
               A:		Dict          =		importlib.metadata.metadata(__lowercase	)
               return True
          except importlib.metadata.PackageNotFoundError:
               return False
def 					SCREAMING_SNAKE_CASE(   )			-> bool:
     return _torch_distributed_available
def 					SCREAMING_SNAKE_CASE(   )			-> Optional[int]:
     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 					SCREAMING_SNAKE_CASE(   )			-> List[str]:
     return importlib.metadata.version('''oneccl_bind_pt'''	)
def 					SCREAMING_SNAKE_CASE(   )			-> Union[str, Any]:
     return _is_package_available('''transformer_engine'''	)
@lru_cache
def 					SCREAMING_SNAKE_CASE(   __lowercase=True	)			-> Union[str, Any]:
     # Due to bugs on the amp series GPUs, we disable torch-xla on them
     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:		Tuple          =		xm.xla_device()
               return True
          except RuntimeError:
               return False
     return _tpu_available
def 					SCREAMING_SNAKE_CASE(   )			-> List[str]:
     return _is_package_available('''deepspeed'''	)
def 					SCREAMING_SNAKE_CASE(   __lowercase=False	)			-> Any:
     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 					SCREAMING_SNAKE_CASE(   )			-> str:
     A:		List[str]          =		_is_package_available('''bitsandbytes'''	)
     if package_exists:
          A:		Union[str, Any]          =		version.parse(importlib.metadata.version('''bitsandbytes'''	)	)
          return compare_versions(__lowercase		,				'''>='''		,				'''0.39.0'''	)
     return False
def 					SCREAMING_SNAKE_CASE(   )			-> str:
     A:		int          =		_is_package_available('''bitsandbytes'''	)
     if package_exists:
          A:		Tuple          =		version.parse(importlib.metadata.version('''bitsandbytes'''	)	)
          return compare_versions(__lowercase		,				'''>='''		,				'''0.37.2'''	)
     return False
def 					SCREAMING_SNAKE_CASE(   )			-> int:
     return _is_package_available('''bitsandbytes'''	)
def 					SCREAMING_SNAKE_CASE(   )			-> Optional[int]:
     if strtobool(os.environ.get('''ACCELERATE_USE_MEGATRON_LM'''		,				'''False'''	)	) == 1:
          A:		Any          =		importlib.util.find_spec('''megatron'''	) is not None
          if package_exists:
               try:
                    A:		str          =		parse(importlib.metadata.version('''megatron-lm'''	)	)
                    return compare_versions(__lowercase		,				'''>='''		,				'''2.2.0'''	)
               except Exception as e:
                    warnings.warn(F"""Parse Megatron version failed. Exception:{e}"""	)
                    return False
def 					SCREAMING_SNAKE_CASE(   )			-> Any:
     return _is_package_available('''safetensors'''	)
def 					SCREAMING_SNAKE_CASE(   )			-> Tuple:
     return _is_package_available('''transformers'''	)
def 					SCREAMING_SNAKE_CASE(   )			-> Union[str, Any]:
     return _is_package_available('''datasets'''	)
def 					SCREAMING_SNAKE_CASE(   )			-> Union[str, Any]:
     return _is_package_available('''aim'''	)
def 					SCREAMING_SNAKE_CASE(   )			-> List[Any]:
     return _is_package_available('''tensorboard'''	) or _is_package_available('''tensorboardX'''	)
def 					SCREAMING_SNAKE_CASE(   )			-> Dict:
     return _is_package_available('''wandb'''	)
def 					SCREAMING_SNAKE_CASE(   )			-> Optional[Any]:
     return _is_package_available('''comet_ml'''	)
def 					SCREAMING_SNAKE_CASE(   )			-> Any:
     return _is_package_available('''boto3'''	)
def 					SCREAMING_SNAKE_CASE(   )			-> str:
     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'''		,				__lowercase	)
          return parse_flag_from_env('''ACCELERATE_ENABLE_RICH'''		,				__lowercase	)
     return False
def 					SCREAMING_SNAKE_CASE(   )			-> int:
     return _is_package_available('''sagemaker'''	)
def 					SCREAMING_SNAKE_CASE(   )			-> Union[str, Any]:
     return _is_package_available('''tqdm'''	)
def 					SCREAMING_SNAKE_CASE(   )			-> str:
     return _is_package_available('''mlflow'''	)
def 					SCREAMING_SNAKE_CASE(   )			-> int:
     return is_torch_version('''>='''		,				'''1.12'''	) and torch.backends.mps.is_available() and torch.backends.mps.is_built()
def 					SCREAMING_SNAKE_CASE(   )			-> str:
     def get_major_and_minor_from_version(__lowercase	):
          return str(version.parse(__lowercase	).major	) + "." + str(version.parse(__lowercase	).minor	)

     A:		int          =		importlib.metadata.version('''torch'''	)
     if importlib.util.find_spec('''intel_extension_for_pytorch'''	) is None:
          return False
     A:		Optional[Any]          =		'''N/A'''
     try:
          A:		List[Any]          =		importlib.metadata.version('''intel_extension_for_pytorch'''	)
     except importlib.metadata.PackageNotFoundError:
          return False
     A:		Optional[int]          =		get_major_and_minor_from_version(_torch_version	)
     A:		List[str]          =		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 					SCREAMING_SNAKE_CASE(   __lowercase=False	)			-> Optional[Any]:
     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:		Tuple          =		torch.npu.device_count()
               return torch.npu.is_available()
          except RuntimeError:
               return False
     return hasattr(__lowercase		,				'''npu'''	) and torch.npu.is_available()


@lru_cache
def 					SCREAMING_SNAKE_CASE(   __lowercase=False	)			-> Tuple:
     if not parse_flag_from_env('''ACCELERATE_USE_XPU'''		,				default=__lowercase	):
          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:		str          =		torch.xpu.device_count()
               return torch.xpu.is_available()
          except RuntimeError:
               return False
     return hasattr(__lowercase		,				'''xpu'''	) and torch.xpu.is_available()






