


"""simple docstring"""







import math
import time

from transformers import Trainer, is_torch_tpu_available
from transformers.trainer_utils import PredictionOutput, speed_metrics


if is_torch_tpu_available(check_device=False):
		import torch_xla.core.xla_model as xm
		import torch_xla.debug.metrics as met




class 				UpperCamelCase_			(				UpperCamelCase):

	"""simple docstring"""





	def __init__(				self    :       str    ,    *UpperCAmelCase__    :       str    ,    UpperCAmelCase__    :       List[str]=None    ,    UpperCAmelCase__    :       str=None    ,    **UpperCAmelCase__    :       Union[str, Any]     ) ->			str:
						super().__init__(*UpperCAmelCase__    ,    **UpperCAmelCase__     )
						__SCREAMING_SNAKE_CASE		     =       eval_examples
						__SCREAMING_SNAKE_CASE		     =       post_process_function




	def 						UpperCAmelCase_   (				self    :       Dict    ,    UpperCAmelCase__    :       Union[str, Any]=None    ,    UpperCAmelCase__    :       str=None    ,    UpperCAmelCase__    :       Dict=None    ,    UpperCAmelCase__    :       str = "eval"     ) ->			str:
						__SCREAMING_SNAKE_CASE		     =       self.eval_dataset if eval_dataset is None else eval_dataset
						__SCREAMING_SNAKE_CASE		     =       self.get_eval_dataloader(UpperCAmelCase__     )
						__SCREAMING_SNAKE_CASE		     =       self.eval_examples if eval_examples is None else eval_examples

						# Temporarily disable metric computation, we will do it in the loop here.
						__SCREAMING_SNAKE_CASE		     =       self.compute_metrics
						__SCREAMING_SNAKE_CASE		     =       None
						__SCREAMING_SNAKE_CASE		     =       self.prediction_loop if self.args.use_legacy_prediction_loop else self.evaluation_loop
						__SCREAMING_SNAKE_CASE		     =       time.time()
						try:
											__SCREAMING_SNAKE_CASE		     =       eval_loop(
											    UpperCAmelCase__    ,    description="Evaluation"    ,    prediction_loss_only=True if compute_metrics is None else None    ,    ignore_keys=UpperCAmelCase__    ,    metric_key_prefix=UpperCAmelCase__    ,    )
						finally:
											__SCREAMING_SNAKE_CASE		     =       compute_metrics
						__SCREAMING_SNAKE_CASE		     =       self.args.eval_batch_size * self.args.world_size
						if F"""{metric_key_prefix}_jit_compilation_time""" in output.metrics:
											start_time += output.metrics[F"""{metric_key_prefix}_jit_compilation_time"""]
						output.metrics.update(
						    speed_metrics(
						        UpperCAmelCase__    ,    UpperCAmelCase__    ,    num_samples=output.num_samples    ,    num_steps=math.ceil(output.num_samples / total_batch_size     )    ,    )     )
						if self.post_process_function is not None and self.compute_metrics is not None and self.args.should_save:
											# Only the main node write the results by default
											__SCREAMING_SNAKE_CASE		     =       self.post_process_function(UpperCAmelCase__    ,    UpperCAmelCase__    ,    output.predictions     )
											__SCREAMING_SNAKE_CASE		     =       self.compute_metrics(UpperCAmelCase__     )

											# Prefix all keys with metric_key_prefix + '_'
											for key in list(metrics.keys()     ):
																if not key.startswith(F"""{metric_key_prefix}_"""     ):
																					__SCREAMING_SNAKE_CASE		     =       metrics.pop(UpperCAmelCase__     )
											metrics.update(output.metrics     )
						else:
											__SCREAMING_SNAKE_CASE		     =       output.metrics

						if self.args.should_log:
											# Only the main node log the results by default
											self.log(UpperCAmelCase__     )

						if self.args.tpu_metrics_debug or self.args.debug:
											# tpu-comment: Logging debug metrics for PyTorch/XLA (compile, execute times, ops, etc.)
											xm.master_print(met.metrics_report()     )

						__SCREAMING_SNAKE_CASE		     =       self.callback_handler.on_evaluate(self.args    ,    self.state    ,    self.control    ,    UpperCAmelCase__     )
						return metrics




	def 						UpperCAmelCase_   (				self    :       Any    ,    UpperCAmelCase__    :       str    ,    UpperCAmelCase__    :       Any    ,    UpperCAmelCase__    :       Dict=None    ,    UpperCAmelCase__    :       str = "test"     ) ->			Dict:
						__SCREAMING_SNAKE_CASE		     =       self.get_test_dataloader(UpperCAmelCase__     )

						# Temporarily disable metric computation, we will do it in the loop here.
						__SCREAMING_SNAKE_CASE		     =       self.compute_metrics
						__SCREAMING_SNAKE_CASE		     =       None
						__SCREAMING_SNAKE_CASE		     =       self.prediction_loop if self.args.use_legacy_prediction_loop else self.evaluation_loop
						__SCREAMING_SNAKE_CASE		     =       time.time()
						try:
											__SCREAMING_SNAKE_CASE		     =       eval_loop(
											    UpperCAmelCase__    ,    description="Prediction"    ,    prediction_loss_only=True if compute_metrics is None else None    ,    ignore_keys=UpperCAmelCase__    ,    metric_key_prefix=UpperCAmelCase__    ,    )
						finally:
											__SCREAMING_SNAKE_CASE		     =       compute_metrics
						__SCREAMING_SNAKE_CASE		     =       self.args.eval_batch_size * self.args.world_size
						if F"""{metric_key_prefix}_jit_compilation_time""" in output.metrics:
											start_time += output.metrics[F"""{metric_key_prefix}_jit_compilation_time"""]
						output.metrics.update(
						    speed_metrics(
						        UpperCAmelCase__    ,    UpperCAmelCase__    ,    num_samples=output.num_samples    ,    num_steps=math.ceil(output.num_samples / total_batch_size     )    ,    )     )

						if self.post_process_function is None or self.compute_metrics is None:
											return output

						__SCREAMING_SNAKE_CASE		     =       self.post_process_function(UpperCAmelCase__    ,    UpperCAmelCase__    ,    output.predictions    ,    "predict"     )
						__SCREAMING_SNAKE_CASE		     =       self.compute_metrics(UpperCAmelCase__     )

						# Prefix all keys with metric_key_prefix + '_'
						for key in list(metrics.keys()     ):
											if not key.startswith(F"""{metric_key_prefix}_"""     ):
																__SCREAMING_SNAKE_CASE		     =       metrics.pop(UpperCAmelCase__     )
						metrics.update(output.metrics     )
						return PredictionOutput(predictions=predictions.predictions    ,    label_ids=predictions.label_ids    ,    metrics=UpperCAmelCase__     )




"""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 				UpperCamelCase_			(				unittest.TestCase):

	"""simple docstring"""





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




	def 						UpperCAmelCase_   (				self    :       Tuple    ,    UpperCAmelCase__    :       int     ) ->			Any:
						__SCREAMING_SNAKE_CASE		     =       F"""{self.env.base_job_name}-{instance_count}-{'ddp' if 'ddp' in self.script else 'smd'}"""
						# distributed data settings
						__SCREAMING_SNAKE_CASE		     =       {"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 						UpperCAmelCase_   (				self    :       Optional[int]    ,    UpperCAmelCase__    :       Optional[int]     ) ->			Optional[Any]:
						TrainingJobAnalytics(UpperCAmelCase__     ).export_csv(F"""{self.env.test_path}/{job_name}_metrics.csv"""     )




	@parameterized.expand([(2,)]     )
	def 						UpperCAmelCase_   (				self    :       Dict    ,    UpperCAmelCase__    :       Optional[Any]     ) ->			List[str]:
						# create estimator
						__SCREAMING_SNAKE_CASE		     =       self.create_estimator(UpperCAmelCase__     )

						# run training
						estimator.fit()

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

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

						# 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__     )
