





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




      def 							lowercase__       (     self     :					str    )							-> Tuple:
            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 							lowercase__       (     self     :					Tuple     ,     __snake_case     :					Union[str, Any]    )							-> Any:
            # configuration for running training on smdistributed Model Parallel
            _lowerCAmelCase					=				{
                """enabled""": True,
                """processes_per_host""": 8,
            }
            _lowerCAmelCase					=				{
                """enabled""": True,
                """parameters""": {
                    """microbatches""": 4,
                    """placement_strategy""": """spread""",
                    """pipeline""": """interleaved""",
                    """optimize""": """speed""",
                    """partitions""": 4,
                    """ddp""": True,
                },
            }

            _lowerCAmelCase					=				{"""smdistributed""": {"""modelparallel""": smp_options}, """mpi""": mpi_options}

            _lowerCAmelCase					=				"""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=__snake_case     ,     instance_type=self.instance_type     ,     debugger_hook_config=__snake_case     ,     hyperparameters={
                    **self.env.hyperparameters,
                    """model_name_or_path""": self.model_name_or_path,
                    """max_steps""": 5_00,
                }     ,     metric_definitions=self.env.metric_definitions     ,     distribution=__snake_case     ,     py_version="""py36"""     ,     )




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




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

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













'''simple docstring'''






# Copyright 2023 The HuggingFace Team. All rights reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

from typing import TYPE_CHECKING

from ...utils import OptionalDependencyNotAvailable, _LazyModule, is_torch_available


A__							:  int    ={
    '''configuration_xmod''': [
        '''XMOD_PRETRAINED_CONFIG_ARCHIVE_MAP''',
        '''XmodConfig''',
        '''XmodOnnxConfig''',
    ],
}

try:
      if not is_torch_available():
            raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
      pass
else:
      A__							:  List[str]    =[
          '''XMOD_PRETRAINED_MODEL_ARCHIVE_LIST''',
          '''XmodForCausalLM''',
          '''XmodForMaskedLM''',
          '''XmodForMultipleChoice''',
          '''XmodForQuestionAnswering''',
          '''XmodForSequenceClassification''',
          '''XmodForTokenClassification''',
          '''XmodModel''',
          '''XmodPreTrainedModel''',
      ]

if TYPE_CHECKING:
      from .configuration_xmod import XMOD_PRETRAINED_CONFIG_ARCHIVE_MAP, XmodConfig, XmodOnnxConfig

      try:
            if not is_torch_available():
                  raise OptionalDependencyNotAvailable()
      except OptionalDependencyNotAvailable:
            pass
      else:
            from .modeling_xmod import (
                XMOD_PRETRAINED_MODEL_ARCHIVE_LIST,
                XmodForCausalLM,
                XmodForMaskedLM,
                XmodForMultipleChoice,
                XmodForQuestionAnswering,
                XmodForSequenceClassification,
                XmodForTokenClassification,
                XmodModel,
                XmodPreTrainedModel,
            )

else:
      import sys

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






