
'''simple docstring'''






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.py''',
            '''model_name_or_path''': '''distilbert-base-cased''',
            '''instance_type''': '''ml.p3.16xlarge''',
            '''results''': {'''train_runtime''': 650, '''eval_accuracy''': 0.7, '''eval_loss''': 0.6},
        },
        {
            '''framework''': '''pytorch''',
            '''script''': '''run_ddp.py''',
            '''model_name_or_path''': '''distilbert-base-cased''',
            '''instance_type''': '''ml.p3.16xlarge''',
            '''results''': {'''train_runtime''': 600, '''eval_accuracy''': 0.7, '''eval_loss''': 0.6},
        },
        {
            '''framework''': '''tensorflow''',
            '''script''': '''run_tf_dist.py''',
            '''model_name_or_path''': '''distilbert-base-cased''',
            '''instance_type''': '''ml.p3.16xlarge''',
            '''results''': {'''train_runtime''': 600, '''eval_accuracy''': 0.6, '''eval_loss''': 0.7},
        },
    ])
class     __snake_case		(   unittest.TestCase):
					def     UpperCAmelCase_							(     self      ):

									"""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=_lowerCAmelCase,    )
									assert hasattr(self,    'env'      )
					def     UpperCAmelCase_							(     self,    A      ):

									"""simple docstring"""


									lowerCamelCase					:			List[str]									= F'''{self.env.base_job_name}-{instance_count}-{'ddp' if 'ddp' in self.script else 'smd'}'''
									# distributed data settings
									lowerCamelCase					:			Optional[int]									= {'smdistributed': {'dataparallel': {'enabled': True}}} if self.script != 'run_ddp.py' else None

									# 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=_lowerCAmelCase,    instance_count=_lowerCAmelCase,    instance_type=self.instance_type,    debugger_hook_config=_lowerCAmelCase,    hyperparameters={**self.env.distributed_hyperparameters, 'model_name_or_path': self.model_name_or_path},    metric_definitions=self.env.metric_definitions,    distribution=_lowerCAmelCase,    py_version='py36',    )
					def     UpperCAmelCase_							(     self,    A      ):

									"""simple docstring"""


									TrainingJobAnalytics(_lowerCAmelCase      ).export_csv(F'''{self.env.test_path}/{job_name}_metrics.csv'''      )




					@parameterized.expand([(2,)]      )
					def     UpperCAmelCase_							(     self,    A      ):

									"""simple docstring"""


									lowerCamelCase					:			Optional[int]									= self.create_estimator(_lowerCAmelCase      )

									# run training
									estimator.fit()

									# result dataframe
									lowerCamelCase					:			List[Any]									= TrainingJobAnalytics(estimator.latest_training_job.name      ).dataframe()

									# extract kpis
									lowerCamelCase					:			Tuple									= list(result_metrics_df[result_metrics_df.metric_name == 'eval_accuracy']['value']      )
									lowerCamelCase					:			Tuple									= list(result_metrics_df[result_metrics_df.metric_name == 'eval_loss']['value']      )
									# get train time from SageMaker job, this includes starting, preprocessing, stopping
									lowerCamelCase					:			Tuple									= (
									    Session().describe_training_job(estimator.latest_training_job.name      ).get('TrainingTimeInSeconds',    99_9999      )
									)

									# 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},    _lowerCAmelCase      )





'''simple docstring'''






import numpy as np





def    UpperCAmelCase							(  UpperCAmelCase__							:							np.array):
				return 1 / (1 + np.exp(-vector))


if __name__ == "__main__":
					import doctest

					doctest.testmod()
