





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       ):




      def     lowercase__   (      self      :			Optional[int]  )    ->				int:
            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     lowercase__   (      self      :			Union[str, Any]      ,      __snake_case      :			Optional[int]  )    ->				Optional[Any]:
            _lowerCAmelCase												=							f"{self.env.base_job_name}-{instance_count}-{'ddp' if 'ddp' in self.script else 'smd'}"
            # distributed data settings
            _lowerCAmelCase												=							{'''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=_lowerCamelCase      ,      instance_count=_lowerCamelCase      ,      instance_type=self.instance_type      ,      debugger_hook_config=_lowerCamelCase      ,      hyperparameters={**self.env.distributed_hyperparameters, """model_name_or_path""": self.model_name_or_path}      ,      metric_definitions=self.env.metric_definitions      ,      distribution=_lowerCamelCase      ,      py_version="""py36"""      ,      )




      def     lowercase__   (      self      :			Union[str, Any]      ,      __snake_case      :			List[str]  )    ->				List[Any]:
            TrainingJobAnalytics(_lowerCamelCase  ).export_csv(f"{self.env.test_path}/{job_name}_metrics.csv"  )




      @parameterized.expand([(2,)]  )
      def     lowercase__   (      self      :			List[Any]      ,      __snake_case      :			Dict  )    ->				Optional[int]:
            # create estimator
            _lowerCAmelCase												=							self.create_estimator(_lowerCamelCase  )

            # run training
            estimator.fit()

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

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













'''simple docstring'''






from typing import TYPE_CHECKING

from ...utils import OptionalDependencyNotAvailable, _LazyModule, is_tokenizers_available


A__    :       int              ={'''tokenization_herbert''': ['''HerbertTokenizer''']}

try:
      if not is_tokenizers_available():
            raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
      pass
else:
      A__    :       Dict              =['''HerbertTokenizerFast''']


if TYPE_CHECKING:
      from .tokenization_herbert import HerbertTokenizer

      try:
            if not is_tokenizers_available():
                  raise OptionalDependencyNotAvailable()
      except OptionalDependencyNotAvailable:
            pass
      else:
            from .tokenization_herbert_fast import HerbertTokenizerFast

else:
      import sys

      A__    :       List[Any]              =_LazyModule(__name__, globals()['''__file__'''], _import_structure, module_spec=__spec__)






