




'''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""": 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     _a						(			unittest.TestCase				):
       def 					A   (			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=_a     ,       )
           assert hasattr(self     ,       '''env'''       )
       def 					A   (			self		:			Optional[Any]     ,       lowercase		:			Dict       ):



           '''simple docstring'''

           UpperCAmelCase     =	f"{self.env.base_job_name}-{instance_count}-{'ddp' if 'ddp' in self.script else 'smd'}"
           # distributed data settings
           UpperCAmelCase     =	{"""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=_a     ,       instance_count=_a     ,       instance_type=self.instance_type     ,       debugger_hook_config=_a     ,       hyperparameters={**self.env.distributed_hyperparameters, '''model_name_or_path''': self.model_name_or_path}     ,       metric_definitions=self.env.metric_definitions     ,       distribution=_a     ,       py_version='''py36'''     ,       )
       def 					A   (			self		:			Tuple     ,       lowercase		:			Tuple       ):



           '''simple docstring'''

           TrainingJobAnalytics(_a       ).export_csv(f"{self.env.test_path}/{job_name}_metrics.csv"       )







       @parameterized.expand([(2,)]       )
       def 					A   (			self		:			Any     ,       lowercase		:			str       ):



           '''simple docstring'''

           UpperCAmelCase     =	self.create_estimator(_a       )

           # 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'''     ,       999_999       )
           )

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










'''simple docstring'''







import warnings

from ...utils import logging
from .image_processing_clip import CLIPImageProcessor


A									=logging.get_logger(__name__)


class     _a						(			__a				):
       def __init__(			self		:			List[str]     ,       *lowercase		:			List[str]     ,       **lowercase		:			Optional[int]       ):



           '''simple docstring'''

           warnings.warn(
               '''The class CLIPFeatureExtractor is deprecated and will be removed in version 5 of Transformers. Please'''
               ''' use CLIPImageProcessor instead.'''     ,       lowercase     ,       )
           super().__init__(*lowercase     ,       **lowercase       )




