
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": 1600, "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": 1600, "eval_accuracy": 0.3, "eval_loss": 1.2},
        },
    ]							)
class       _lowerCAmelCase  (		unittest.TestCase							):

 """simple docstring"""

 def      __lowerCAmelCase  (						self			:			str   ):






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

 def      __lowerCAmelCase  (						self			:			Tuple					, SCREAMING_SNAKE_CASE__			:			Any   ):






      """simple docstring"""



      UpperCamelCase						    =					{
          'enabled': True,
          'processes_per_host': 8,
      }
      UpperCamelCase						    =					{
          'enabled': True,
          'parameters': {
              'microbatches': 4,
              'placement_strategy': 'spread',
              'pipeline': 'interleaved',
              'optimize': 'speed',
              'partitions': 4,
              'ddp': True,
          },
      }

      UpperCamelCase						    =					{'smdistributed': {'modelparallel': smp_options}, 'mpi': mpi_options}

      UpperCamelCase						    =					'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=SCREAMING_SNAKE_CASE__					, instance_type=self.instance_type					, debugger_hook_config=SCREAMING_SNAKE_CASE__					, hyperparameters={
              **self.env.hyperparameters,
              'model_name_or_path': self.model_name_or_path,
              'max_steps': 5_00,
          }					, metric_definitions=self.env.metric_definitions					, distribution=SCREAMING_SNAKE_CASE__					, py_version='py36'					, )

 def      __lowerCAmelCase  (						self			:			Union[str, Any]					, SCREAMING_SNAKE_CASE__			:			Optional[int]   ):






      """simple docstring"""



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






 @parameterized.expand([(1,)]   )
 def      __lowerCAmelCase  (						self			:			Tuple					, SCREAMING_SNAKE_CASE__			:			Union[str, Any]   ):






      """simple docstring"""



      UpperCamelCase						    =					self.create_estimator(SCREAMING_SNAKE_CASE__   )

      # run training
      estimator.fit()

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

      # extract kpis
      UpperCamelCase						    =					list(result_metrics_df[result_metrics_df.metric_name == 'eval_accuracy']['value']   )
      UpperCamelCase						    =					list(result_metrics_df[result_metrics_df.metric_name == 'eval_loss']['value']   )
      # get train time from SageMaker job, this includes starting, preprocessing, stopping
      UpperCamelCase						    =					(
          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__   )




import json
import os
import tempfile

from transformers.testing_utils import check_json_file_has_correct_format
class       _lowerCAmelCase  :

 """simple docstring"""
 SCREAMING_SNAKE_CASE_				:					Optional[Any]       						=None

 def      __lowerCAmelCase  (						self			:			Union[str, Any]   ):






      """simple docstring"""



      UpperCamelCase						    =					self.feature_extraction_class(**self.feat_extract_dict   )
      UpperCamelCase						    =					json.loads(feat_extract.to_json_string()   )
      for key, value in self.feat_extract_dict.items():
           self.assertEqual(obj[key]					, SCREAMING_SNAKE_CASE__   )

 def      __lowerCAmelCase  (						self			:			Tuple   ):






      """simple docstring"""



      UpperCamelCase						    =					self.feature_extraction_class(**self.feat_extract_dict   )

      with tempfile.TemporaryDirectory() as tmpdirname:
           UpperCamelCase						    =					os.path.join(SCREAMING_SNAKE_CASE__					, 'feat_extract.json'   )
           feat_extract_first.to_json_file(SCREAMING_SNAKE_CASE__   )
           UpperCamelCase						    =					self.feature_extraction_class.from_json_file(SCREAMING_SNAKE_CASE__   )

      self.assertEqual(feat_extract_second.to_dict()					, feat_extract_first.to_dict()   )

 def      __lowerCAmelCase  (						self			:			Tuple   ):






      """simple docstring"""



      UpperCamelCase						    =					self.feature_extraction_class(**self.feat_extract_dict   )

      with tempfile.TemporaryDirectory() as tmpdirname:
           UpperCamelCase						    =					feat_extract_first.save_pretrained(SCREAMING_SNAKE_CASE__   )[0]
           check_json_file_has_correct_format(SCREAMING_SNAKE_CASE__   )
           UpperCamelCase						    =					self.feature_extraction_class.from_pretrained(SCREAMING_SNAKE_CASE__   )

      self.assertEqual(feat_extract_second.to_dict()					, feat_extract_first.to_dict()   )






 def      __lowerCAmelCase  (						self			:			Union[str, Any]   ):






      """simple docstring"""



      UpperCamelCase						    =					self.feature_extraction_class()
      self.assertIsNotNone(SCREAMING_SNAKE_CASE__   )


