



'''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 	lowerCamelCase_    ( unittest.TestCase						):






						'''simple docstring'''







						def 			SCREAMING_SNAKE_CASE__		(	self      :			Dict )      ->				List[Any]:

									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=__SCREAMING_SNAKE_CASE	,	)
									assert hasattr(self	,	"env" )







						def 			SCREAMING_SNAKE_CASE__		(	self      :			Optional[int]	,	__lowerCamelCase      :			str )      ->				List[str]:
									A							:					Optional[int]              =						F"""{self.env.base_job_name}-{instance_count}-{"ddp" if "ddp" in self.script else "smd"}"""
									# distributed data settings
									A							:					Tuple              =						{"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=__SCREAMING_SNAKE_CASE	,	instance_count=__SCREAMING_SNAKE_CASE	,	instance_type=self.instance_type	,	debugger_hook_config=__SCREAMING_SNAKE_CASE	,	hyperparameters={**self.env.distributed_hyperparameters, "model_name_or_path": self.model_name_or_path}	,	metric_definitions=self.env.metric_definitions	,	distribution=__SCREAMING_SNAKE_CASE	,	py_version="py36"	,	)







						def 			SCREAMING_SNAKE_CASE__		(	self      :			Tuple	,	__lowerCamelCase      :			Optional[Any] )      ->				Union[str, Any]:
									TrainingJobAnalytics(__SCREAMING_SNAKE_CASE ).export_csv(F"""{self.env.test_path}/{job_name}_metrics.csv""" )







						@parameterized.expand([(2,)] )
						def 			SCREAMING_SNAKE_CASE__		(	self      :			Optional[int]	,	__lowerCamelCase      :			Any )      ->				Any:
									A							:					Dict              =						self.create_estimator(__SCREAMING_SNAKE_CASE )

									# run training
									estimator.fit()

									# result dataframe
									A							:					str              =						TrainingJobAnalytics(estimator.latest_training_job.name ).dataframe()

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

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




from dataclasses import dataclass, field
from typing import ClassVar, Dict

from ..features import Features, Sequence, Value
from .base import TaskTemplate

@dataclass(frozen=_A						)
class 	lowerCamelCase_    ( _A						):






						'''simple docstring'''
						# `task` is not a ClassVar since we want it to be part of the `asdict` output for JSON serialization
						a__   =  field(default="question-answering-extractive"						,metadata={"include_in_asdict_even_if_is_default": True}						)
						a__   =  Features({"question": Value("string"						), "context": Value("string"						)}						)
						a__   =  Features(
						    {
						        "answers": Sequence(
						            {
						                "text": Value("string"						),
						                "answer_start": Value("int32"						),
						            }						)
						    }						)
						a__   =  "question"
						a__   =  "context"
						a__   =  "answers"







						@property
						def 			SCREAMING_SNAKE_CASE__		(	self      :			Union[str, Any] )      ->				Dict[str, str]:
									return {self.question_column: "question", self.context_column: "context", self.answers_column: "answers"}