



import importlib.util
import json
import os
import warnings
from dataclasses import dataclass, field

import torch

from ..training_args import TrainingArguments
from ..utils import cached_property, is_sagemaker_dp_enabled, logging


lowerCamelCase__			= logging.get_logger(__name__)

def   lowerCAmelCase__						(							):
				"""simple docstring"""







				__a       =		os.getenv("""SM_HP_MP_PARAMETERS"""				,     """{}"""    )
				try:
								# Parse it and check the field "partitions" is included, it is required for model parallel.
								__a       =		json.loads(_SCREAMING_SNAKE_CASE    )
								if "partitions" not in smp_options:
												return False
				except json.JSONDecodeError:
								return False

				# Get the sagemaker specific framework parameters from mpi_options variable.
				__a       =		os.getenv("""SM_FRAMEWORK_PARAMS"""				,     """{}"""    )
				try:
								# Parse it and check the field "sagemaker_distributed_dataparallel_enabled".
								__a       =		json.loads(_SCREAMING_SNAKE_CASE    )
								if not mpi_options.get("""sagemaker_mpi_enabled"""				,     _SCREAMING_SNAKE_CASE    ):
												return False
				except json.JSONDecodeError:
								return False
				# Lastly, check if the `smdistributed` module is present.
				return importlib.util.find_spec("""smdistributed"""    ) is not None


if is_sagemaker_model_parallel_available():
						import smdistributed.modelparallel.torch as smp

						smp.init()






@dataclass
class    SCREAMING_SNAKE_CASE			(							lowerCamelCase__					):
							__lowerCamelCase	:	str    =field(
							    default=''			,      metadata={'help': 'Used by the SageMaker launcher to send mp-specific args. Ignored in SageMakerTrainer'}			,      )





							def 						UpperCamelCase_					(   self		:		str	):

											'''simple docstring'''
											super().__post_init__()
											warnings.warn(
											    """`SageMakerTrainingArguments` is deprecated and will be removed in v5 of Transformers. You can use """
											    """`TrainingArguments` instead."""      ,				__lowercase      ,				)





							@cached_property
							def 						UpperCamelCase_					(   self		:		List[Any]	):

											'''simple docstring'''
											logger.info("""PyTorch: setting up devices"""	)
											if torch.distributed.is_available() and torch.distributed.is_initialized() and self.local_rank == -1:
															logger.warning(
															    """torch.distributed process group is initialized, but local_rank == -1. """
															    """In order to use Torch DDP, launch your script with `python -m torch.distributed.launch"""	)
											if self.no_cuda:
															__a       =		torch.device("""cpu"""	)
															__a       =		0
											elif is_sagemaker_model_parallel_available():
															__a       =		smp.local_rank()
															__a       =		torch.device("""cuda"""      ,				__lowercase	)
															__a       =		1
											elif is_sagemaker_dp_enabled():
															import smdistributed.dataparallel.torch.torch_smddp  # noqa: F401

															torch.distributed.init_process_group(backend="""smddp"""      ,				timeout=self.ddp_timeout_delta	)
															__a       =		int(os.getenv("""SMDATAPARALLEL_LOCAL_RANK"""	)	)
															__a       =		torch.device("""cuda"""      ,				self.local_rank	)
															__a       =		1
											elif self.local_rank == -1:
															# if n_gpu is > 1 we'll use nn.DataParallel.
															# If you only want to use a specific subset of GPUs use `CUDA_VISIBLE_DEVICES=0`
															# Explicitly set CUDA to the first (index 0) CUDA device, otherwise `set_device` will
															# trigger an error that a device index is missing. Index 0 takes into account the
															# GPUs available in the environment, so `CUDA_VISIBLE_DEVICES=1,2` with `cuda:0`
															# will use the first GPU in that env, i.e. GPU#1
															__a       =		torch.device("""cuda:0""" if torch.cuda.is_available() else """cpu"""	)
															# Sometimes the line in the postinit has not been run before we end up here, so just checking we're not at
															# the default value.
															__a       =		torch.cuda.device_count()
											else:
															# Here, we'll use torch.distributed.
															# Initializes the distributed backend which will take care of synchronizing nodes/GPUs
															if not torch.distributed.is_initialized():
																			torch.distributed.init_process_group(backend="""nccl"""      ,				timeout=self.ddp_timeout_delta	)
															__a       =		torch.device("""cuda"""      ,				self.local_rank	)
															__a       =		1

											if device.type == "cuda":
															torch.cuda.set_device(__lowercase	)

											return device





							@property
							def 						UpperCamelCase_					(   self		:		str	):

											'''simple docstring'''
											if is_sagemaker_model_parallel_available():
															return smp.dp_size()

											return super().world_size





							@property
							def 						UpperCamelCase_					(   self		:		Optional[Any]	):

											'''simple docstring'''
											return not is_sagemaker_model_parallel_available()







							@property
							def 						UpperCamelCase_					(   self		:		Optional[int]	):

											'''simple docstring'''
											return False





import unittest
from dataclasses import dataclass

import pytest

from accelerate.commands.config.config_args import SageMakerConfig
from accelerate.utils import ComputeEnvironment
from accelerate.utils.launch import _convert_nargs_to_dict






@dataclass
class    SCREAMING_SNAKE_CASE			(							lowerCamelCase__					):
							__lowerCamelCase	:	List[Any]    =ComputeEnvironment.AMAZON_SAGEMAKER
							__lowerCamelCase	:	Optional[int]    =True
							__lowerCamelCase	:	List[Any]    ='ml.p3.2xlarge'
							__lowerCamelCase	:	Optional[int]    ='accelerate_sagemaker_execution_role'
							__lowerCamelCase	:	Optional[Any]    ='hf-sm'
							__lowerCamelCase	:	Any    ='us-east-1'
							__lowerCamelCase	:	Any    =1
							__lowerCamelCase	:	List[str]    ='accelerate-sagemaker-1'
							__lowerCamelCase	:	Union[str, Any]    ='1.6'
							__lowerCamelCase	:	str    ='4.4'
							__lowerCamelCase	:	List[str]    ='train.py'
							__lowerCamelCase	:	List[Any]    =[
							    '--model_name_or_path',
							    'bert',
							    '--do_train',
							    'False',
							    '--epochs',
							    '3',
							    '--learning_rate',
							    '5e-5',
							    '--max_steps',
							    '50.5',
							]
							__lowerCamelCase	:	Tuple    =[
							    '--model_name_or_path',
							    'bert',
							    '--do_train',
							    '--do_test',
							    'False',
							    '--do_predict',
							    '--epochs',
							    '3',
							    '--learning_rate',
							    '5e-5',
							    '--max_steps',
							    '50.5',
							]






class    SCREAMING_SNAKE_CASE			(							unittest.TestCase					):





							def 						UpperCamelCase_					(   self		:		Union[str, Any]	):

											'''simple docstring'''
											# If no defaults are changed, `to_kwargs` returns an empty dict.
											__a       =		_convert_nargs_to_dict(MockLaunchConfig.success_training_script_args	)
											assert isinstance(converted_args["""model_name_or_path"""]      ,				__lowercase	)
											assert isinstance(converted_args["""do_train"""]      ,				__lowercase	)
											assert isinstance(converted_args["""epochs"""]      ,				__lowercase	)
											assert isinstance(converted_args["""learning_rate"""]      ,				__lowercase	)
											assert isinstance(converted_args["""max_steps"""]      ,				__lowercase	)

											with pytest.raises(__lowercase	):
															_convert_nargs_to_dict(MockLaunchConfig.fail_training_script_args	)
