


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

   '''simple docstring'''







   def   _snake_case    (  self  : List[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=_lowerCamelCase       ,  )
         assert hasattr(self       ,  """env"""							)

   def   _snake_case    (  self  : Tuple       ,  _lowerCamelCase  : Dict							):






         '''simple docstring'''



         __lowerCamelCase     :      str       						=						{
             """enabled""": True,
             """processes_per_host""": 8,
         }
         __lowerCamelCase     :      List[Any]       						=						{
             """enabled""": True,
             """parameters""": {
                 """microbatches""": 4,
                 """placement_strategy""": """spread""",
                 """pipeline""": """interleaved""",
                 """optimize""": """speed""",
                 """partitions""": 4,
                 """ddp""": True,
             },
         }

         __lowerCamelCase     :      int       						=						{"""smdistributed""": {"""modelparallel""": smp_options}, """mpi""": mpi_options}

         __lowerCamelCase     :      List[str]       						=						"""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=_lowerCamelCase       ,  instance_type=self.instance_type       ,  debugger_hook_config=_lowerCamelCase       ,  hyperparameters={
                 **self.env.hyperparameters,
                 """model_name_or_path""": self.model_name_or_path,
                 """max_steps""": 5_0_0,
             }       ,  metric_definitions=self.env.metric_definitions       ,  distribution=_lowerCamelCase       ,  py_version="""py36"""       ,  )

   def   _snake_case    (  self  : Optional[int]       ,  _lowerCamelCase  : int							):






         '''simple docstring'''



         TrainingJobAnalytics(_lowerCamelCase							).export_csv(F"""{self.env.test_path}/{job_name}_metrics.csv"""							)


   @parameterized.expand([(1,)]							)
   def   _snake_case    (  self  : List[str]       ,  _lowerCamelCase  : Optional[int]							):






         '''simple docstring'''



         __lowerCamelCase     :      Tuple       						=						self.create_estimator(_lowerCamelCase							)

         # run training
         estimator.fit()

         # result dataframe
         __lowerCamelCase     :      List[str]       						=						TrainingJobAnalytics(estimator.latest_training_job.name							).dataframe()

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







from typing import Any





def   _UpperCAmelCase       (     UpperCAmelCase      :			list      ):
      """simple docstring"""




      if not input_list:
            return []
      __lowerCamelCase     :      int       						=						[input_list.count(UpperCAmelCase      ) for value in input_list]
      __lowerCamelCase     :      Tuple       						=						max(UpperCAmelCase      )  # Gets the maximum count in the input list.
      # Gets values of modes
      return sorted({input_list[i] for i, value in enumerate(UpperCAmelCase      ) if value == y}      )


if __name__ == "__main__":
    import doctest

    doctest.testmod()



