

'''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': 6_50, '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': 6_00, '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': 6_00, 'eval_accuracy': 0.6, 'eval_loss': 0.7},
        },
    ]			)
class 		__SCREAMING_SNAKE_CASE       ( unittest.TestCase			):







				def    lowerCamelCase_  (					self    :		Union[str, Any]      ):
										'''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=UpperCAmelCase__		,				)
										assert hasattr(self		,				'''env'''      )







				def    lowerCamelCase_  (					self    :		Optional[Any]		,				UpperCAmelCase__    :		int      ):
										'''simple docstring'''
										lowercase   :       str								=F'''{self.env.base_job_name}-{instance_count}-{"ddp" if "ddp" in self.script else "smd"}'''
										# distributed data settings
										lowercase   :       str								={'''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=UpperCAmelCase__		,				instance_count=UpperCAmelCase__		,				instance_type=self.instance_type		,				debugger_hook_config=UpperCAmelCase__		,				hyperparameters={**self.env.distributed_hyperparameters, '''model_name_or_path''': self.model_name_or_path}		,				metric_definitions=self.env.metric_definitions		,				distribution=UpperCAmelCase__		,				py_version='''py36'''		,				)







				def    lowerCamelCase_  (					self    :		Optional[int]		,				UpperCAmelCase__    :		Dict      ):
										'''simple docstring'''
										TrainingJobAnalytics(UpperCAmelCase__      ).export_csv(F'''{self.env.test_path}/{job_name}_metrics.csv'''      )







				@parameterized.expand([(2,)]      )
				def    lowerCamelCase_  (					self    :		str		,				UpperCAmelCase__    :		int      ):
										'''simple docstring'''
										lowercase   :       Dict								=self.create_estimator(UpperCAmelCase__      )

										# run training
										estimator.fit()

										# result dataframe
										lowercase   :       Any								=TrainingJobAnalytics(estimator.latest_training_job.name      ).dataframe()

										# extract kpis
										lowercase   :       Tuple								=list(result_metrics_df[result_metrics_df.metric_name == '''eval_accuracy''']['''value''']      )
										lowercase   :       Optional[int]								=list(result_metrics_df[result_metrics_df.metric_name == '''eval_loss''']['''value''']      )
										# get train time from SageMaker job, this includes starting, preprocessing, stopping
										lowercase   :       Dict								=(
										    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}		,				UpperCAmelCase__      )








'''simple docstring'''





# coding=utf-8
# Copyright 2023 The HuggingFace Inc. team.
#
# 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.

# this script dumps information about the environment

import os
import platform
import sys


UpperCamelCase_							=					"""3"""

print("""Python version:""", sys.version)

print("""OS platform:""", platform.platform())
print("""OS architecture:""", platform.machine())

try:
							import torch

							print("""Torch version:""", torch.__version__)
							print("""Cuda available:""", torch.cuda.is_available())
							print("""Cuda version:""", torch.version.cuda)
							print("""CuDNN version:""", torch.backends.cudnn.version())
							print("""Number of GPUs available:""", torch.cuda.device_count())
except ImportError:
							print("""Torch version:""", None)

try:
							import transformers

							print("""transformers version:""", transformers.__version__)
except ImportError:
							print("""transformers version:""", None)





