


'''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": 6_5_0, "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": 6_0_0, "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": 6_0_0, "eval_accuracy": 0.6, "eval_loss": 0.7},
        },
    ]	)
class 	UpperCamelCase__					( unittest.TestCase	):



     """simple docstring"""






     def a	( self   ):

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




     def a	( self      ,	snake_case__   ):

            '''simple docstring'''

            _lowerCAmelCase    :							List[str]   			=       F'{self.env.base_job_name}-{instance_count}-{"ddp" if "ddp" in self.script else "smd"}'
            # distributed data settings
            _lowerCAmelCase    :							str   			=       {'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=snake_case__      ,	instance_count=snake_case__      ,	instance_type=self.instance_type      ,	debugger_hook_config=snake_case__      ,	hyperparameters={**self.env.distributed_hyperparameters, 'model_name_or_path': self.model_name_or_path}      ,	metric_definitions=self.env.metric_definitions      ,	distribution=snake_case__      ,	py_version='py36'      ,	)




     def a	( self      ,	snake_case__   ):

            '''simple docstring'''

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





     @parameterized.expand([(2,)]   )
     def a	( self      ,	snake_case__   ):

            '''simple docstring'''

            _lowerCAmelCase    :							Dict   			=       self.create_estimator(snake_case__   )

            # run training
            estimator.fit()

            # result dataframe
            _lowerCAmelCase    :							Any   			=       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    :							Dict   			=       (
                Session().describe_training_job(estimator.latest_training_job.name   ).get('TrainingTimeInSeconds'      ,	99_9999   )
            )

            # 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}      ,	snake_case__   )





'''simple docstring'''







from typing import TYPE_CHECKING

from ....utils import OptionalDependencyNotAvailable, _LazyModule, is_torch_available


lowerCAmelCase    :     Optional[int]					=						{
    """configuration_mctct""": ["""MCTCT_PRETRAINED_CONFIG_ARCHIVE_MAP""", """MCTCTConfig"""],
    """feature_extraction_mctct""": ["""MCTCTFeatureExtractor"""],
    """processing_mctct""": ["""MCTCTProcessor"""],
}


try:
       if not is_torch_available():
              raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
       pass
else:
       lowerCAmelCase    :     List[str]					=						[
           """MCTCT_PRETRAINED_MODEL_ARCHIVE_LIST""",
           """MCTCTForCTC""",
           """MCTCTModel""",
           """MCTCTPreTrainedModel""",
       ]


if TYPE_CHECKING:
       from .configuration_mctct import MCTCT_PRETRAINED_CONFIG_ARCHIVE_MAP, MCTCTConfig
       from .feature_extraction_mctct import MCTCTFeatureExtractor
       from .processing_mctct import MCTCTProcessor

       try:
              if not is_torch_available():
                     raise OptionalDependencyNotAvailable()
       except OptionalDependencyNotAvailable:
              pass
       else:
              from .modeling_mctct import MCTCT_PRETRAINED_MODEL_ARCHIVE_LIST, MCTCTForCTC, MCTCTModel, MCTCTPreTrainedModel

else:
       import sys

       lowerCAmelCase    :     Tuple					=						_LazyModule(__name__, globals()["""__file__"""], _import_structure, module_spec=__spec__)

