


'''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   a__(	self    :				List[Any]   )->  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=lowerCAmelCase   ,					)
										assert hasattr(self   ,					'''env'''   )




				def   a__(	self    :				int   ,					lowerCAmelCase    :				Any   )->  Any:

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




				def   a__(	self    :				Union[str, Any]   ,					lowerCAmelCase    :				Union[str, Any]   )->  Any:

										"""simple docstring"""




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





				@parameterized.expand([(1,)]   )
				def   a__(	self    :				str   ,					lowerCAmelCase    :				str   )->  Optional[Any]:

										"""simple docstring"""




										UpperCAmelCase         =			self.create_estimator(lowerCAmelCase   )

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









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


_lowercase    :    int			     =  {
    """configuration_xmod""": [
        """XMOD_PRETRAINED_CONFIG_ARCHIVE_MAP""",
        """XmodConfig""",
        """XmodOnnxConfig""",
    ],
}

try:
	if not is_torch_available():
		raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
	pass
else:
	_lowercase    :    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

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





