






'''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_model_parallelism.py""",
            """model_name_or_path""": """roberta-large""",
            """instance_type""": """ml.p3dn.24xlarge""",
            """results""": {"""train_runtime""": 16_00, """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""": 16_00, """eval_accuracy""": 0.3, """eval_loss""": 1.2},
        },
    ]       )
class        a_		(    unittest.TestCase       ):

  def 						A__		(   self   )						->   Dict:



        """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 						A__		(   self    ,	_SCREAMING_SNAKE_CASE   )						->   Dict:



        """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""": 500,
            }    ,	metric_definitions=self.env.metric_definitions    ,	distribution=_SCREAMING_SNAKE_CASE    ,	py_version="""py36"""    ,	)

  def 						A__		(   self    ,	_SCREAMING_SNAKE_CASE   )						->   Optional[Any]:



        """simple docstring"""






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


  @parameterized.expand([(1,)]   )
  def 						A__		(   self    ,	_SCREAMING_SNAKE_CASE   )						->   List[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"""    ,	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}    ,	_SCREAMING_SNAKE_CASE   )








'''simple docstring'''

def 					lowercase__				(   __UpperCamelCase					)->				str:
      if not all(char in """01""" for char in bin_string					):
            raise ValueError("""Non-binary value was passed to the function"""					)
      if not bin_string:
            raise ValueError("""Empty string was passed to the function"""					)
      UpperCamelCase       						=      """"""
      while len(__UpperCamelCase					) % 3 != 0:
            UpperCamelCase       						=      """0""" + bin_string
      UpperCamelCase       						=      [
          bin_string[index : index + 3]
          for index in range(len(__UpperCamelCase					)					)
          if index % 3 == 0
      ]
      for bin_group in bin_string_in_3_list:
            UpperCamelCase       						=      0
            for index, val in enumerate(__UpperCamelCase					):
                  oct_val += int(2 ** (2 - index) * int(__UpperCamelCase					)					)
            oct_string += str(__UpperCamelCase					)
      return oct_string


if __name__ == "__main__":
     from doctest import testmod

     testmod()
