



import json
import os
import subprocess
import unittest
from ast import literal_eval

import pytest
from parameterized import parameterized, parameterized_class

from . import is_sagemaker_available


if is_sagemaker_available():
						from sagemaker import Session, TrainingJobAnalytics
						from sagemaker.huggingface import HuggingFace






@pytest.mark.skipif(
    literal_eval(os.getenv('TEST_SAGEMAKER'			,      'False'					)					) is not True			,      reason='Skipping test because should only be run when releasing minor transformers version'			,      )
@pytest.mark.usefixtures('sm_env'					)
@parameterized_class(
    [
        {
            'framework': 'pytorch',
            'script': 'run_glue_model_parallelism.py',
            'model_name_or_path': 'roberta-large',
            'instance_type': 'ml.p3dn.24xlarge',
            'results': {'train_runtime': 1_600, 'eval_accuracy': 0.3, 'eval_loss': 1.2},
        },
        {
            'framework': 'pytorch',
            'script': 'run_glue.py',
            'model_name_or_path': 'roberta-large',
            'instance_type': 'ml.p3dn.24xlarge',
            'results': {'train_runtime': 1_600, 'eval_accuracy': 0.3, 'eval_loss': 1.2},
        },
    ]					)
class    SCREAMING_SNAKE_CASE			(							unittest.TestCase					):





							def 						UpperCamelCase_					(   self		:		Dict	):

											'''simple docstring'''
											if self.framework == "pytorch":
															subprocess.run(
															    F"cp ./examples/pytorch/text-classification/run_glue.py {self.env.test_path}/run_glue.py".split()      ,				encoding="""utf-8"""      ,				check=__lowercase      ,				)
											assert hasattr(self      ,				"""env"""	)





							def 						UpperCamelCase_					(   self		:		int      ,				__lowercase		:		Any	):

											'''simple docstring'''
											# configuration for running training on smdistributed Model Parallel
											__a       =		{
											    """enabled""": True,
											    """processes_per_host""": 8,
											}
											__a       =		{
											    """enabled""": True,
											    """parameters""": {
											        """microbatches""": 4,
											        """placement_strategy""": """spread""",
											        """pipeline""": """interleaved""",
											        """optimize""": """speed""",
											        """partitions""": 4,
											        """ddp""": True,
											    },
											}

											__a       =		{"""smdistributed""": {"""modelparallel""": smp_options}, """mpi""": mpi_options}

											__a       =		"""trainer""" if self.script == """run_glue.py""" else """smtrainer"""
											# creates estimator
											return HuggingFace(
											    entry_point=self.script      ,				source_dir=self.env.test_path      ,				role=self.env.role      ,				image_uri=self.env.image_uri      ,				base_job_name=F"{self.env.base_job_name}-{instance_count}-smp-{name_extension}"      ,				instance_count=__lowercase      ,				instance_type=self.instance_type      ,				debugger_hook_config=__lowercase      ,				hyperparameters={
											        **self.env.hyperparameters,
											        """model_name_or_path""": self.model_name_or_path,
											        """max_steps""": 500,
											    }      ,				metric_definitions=self.env.metric_definitions      ,				distribution=__lowercase      ,				py_version="""py36"""      ,				)





							def 						UpperCamelCase_					(   self		:		Optional[Any]      ,				__lowercase		:		Optional[int]	):

											'''simple docstring'''
											TrainingJobAnalytics(__lowercase	).export_csv(F"{self.env.test_path}/{job_name}_metrics.csv"	)







							@parameterized.expand([(1,)]	)
							def 						UpperCamelCase_					(   self		:		Union[str, Any]      ,				__lowercase		:		Union[str, Any]	):

											'''simple docstring'''
											# create estimator
											__a       =		self.create_estimator(__lowercase	)

											# run training
											estimator.fit()

											# result dataframe
											__a       =		TrainingJobAnalytics(estimator.latest_training_job.name	).dataframe()

											# extract kpis
											__a       =		list(result_metrics_df[result_metrics_df.metric_name == """eval_accuracy"""]["""value"""]	)
											__a       =		list(result_metrics_df[result_metrics_df.metric_name == """eval_loss"""]["""value"""]	)
											# get train time from SageMaker job, this includes starting, preprocessing, stopping
											__a       =		(
											    Session().describe_training_job(estimator.latest_training_job.name	).get("""TrainingTimeInSeconds"""      ,				999999	)
											)

											# assert kpis
											assert train_runtime <= self.results["train_runtime"]
											assert all(t >= self.results["""eval_accuracy"""] for t in eval_accuracy	)
											assert all(t <= self.results["""eval_loss"""] for t in eval_loss	)

											# dump tests result into json file to share in PR
											with open(F"{estimator.latest_training_job.name}.json"      ,				"""w"""	) as outfile:
															json.dump({"""train_time""": train_runtime, """eval_accuracy""": eval_accuracy, """eval_loss""": eval_loss}      ,				__lowercase	)





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	)
