
import inspect
import tempfile
import unittest

from huggingface_hub import hf_hub_download

from transformers import is_torch_available
from transformers.testing_utils import is_flaky, require_torch, slow, torch_device

from ...test_configuration_common import ConfigTester
from ...test_modeling_common import ModelTesterMixin, floats_tensor, ids_tensor
from ...test_pipeline_mixin import PipelineTesterMixin


SCREAMING_SNAKE_CASE__   :			Tuple												=    1E-4

if is_torch_available():
					import torch

					from transformers import AutoformerConfig, AutoformerForPrediction, AutoformerModel
					from transformers.models.autoformer.modeling_autoformer import AutoformerDecoder, AutoformerEncoder

@require_torch
class    lowerCAmelCase__     :






		def __init__(		self		:							List[str]							,			SCREAMING_SNAKE_CASE__		:							List[str]							,			SCREAMING_SNAKE_CASE__		:							Optional[int]=16							,			SCREAMING_SNAKE_CASE__		:							Tuple=13							,			SCREAMING_SNAKE_CASE__		:							str=7							,			SCREAMING_SNAKE_CASE__		:							Union[str, Any]=14							,			SCREAMING_SNAKE_CASE__		:							Optional[Any]=10							,			SCREAMING_SNAKE_CASE__		:							Union[str, Any]=19							,			SCREAMING_SNAKE_CASE__		:							Union[str, Any]=5							,			SCREAMING_SNAKE_CASE__		:							Optional[int]=4							,			SCREAMING_SNAKE_CASE__		:							int=True							,			SCREAMING_SNAKE_CASE__		:							Union[str, Any]=16							,			SCREAMING_SNAKE_CASE__		:							Any=2							,			SCREAMING_SNAKE_CASE__		:							Dict=4							,			SCREAMING_SNAKE_CASE__		:							Any=4							,			SCREAMING_SNAKE_CASE__		:							Any="gelu"							,			SCREAMING_SNAKE_CASE__		:							List[str]=0.1							,			SCREAMING_SNAKE_CASE__		:							int=0.1							,			SCREAMING_SNAKE_CASE__		:							Dict=[1, 2, 3, 4, 5]							,			SCREAMING_SNAKE_CASE__		:							List[str]=25							,			SCREAMING_SNAKE_CASE__		:							List[str]=5							,			)     ->							List[Any]:
						__lowerCamelCase    					=	d_model
						__lowerCamelCase    					=	parent
						__lowerCamelCase    					=	batch_size
						__lowerCamelCase    					=	prediction_length
						__lowerCamelCase    					=	context_length
						__lowerCamelCase    					=	cardinality
						__lowerCamelCase    					=	num_time_features
						__lowerCamelCase    					=	lags_sequence
						__lowerCamelCase    					=	embedding_dimension
						__lowerCamelCase    					=	is_training
						__lowerCamelCase    					=	hidden_size
						__lowerCamelCase    					=	num_hidden_layers
						__lowerCamelCase    					=	num_attention_heads
						__lowerCamelCase    					=	intermediate_size
						__lowerCamelCase    					=	hidden_act
						__lowerCamelCase    					=	hidden_dropout_prob
						__lowerCamelCase    					=	attention_probs_dropout_prob

						__lowerCamelCase    					=	context_length
						__lowerCamelCase    					=	prediction_length + label_length
						__lowerCamelCase    					=	label_length

						__lowerCamelCase    					=	moving_average
						__lowerCamelCase    					=	autocorrelation_factor






		def  __A       (		self		:							str  )     ->							Union[str, Any]:
						return AutoformerConfig(
						    d_model=self.d_model							,			encoder_layers=self.num_hidden_layers							,			decoder_layers=self.num_hidden_layers							,			encoder_attention_heads=self.num_attention_heads							,			decoder_attention_heads=self.num_attention_heads							,			encoder_ffn_dim=self.intermediate_size							,			decoder_ffn_dim=self.intermediate_size							,			dropout=self.hidden_dropout_prob							,			attention_dropout=self.attention_probs_dropout_prob							,			prediction_length=self.prediction_length							,			context_length=self.context_length							,			label_length=self.label_length							,			lags_sequence=self.lags_sequence							,			num_time_features=self.num_time_features							,			num_static_categorical_features=1							,			cardinality=[self.cardinality]							,			embedding_dimension=[self.embedding_dimension]							,			moving_average=self.moving_average							,			)






		def  __A       (		self		:							List[Any]							,			SCREAMING_SNAKE_CASE__		:							Dict  )     ->							Any:
						__lowerCamelCase    					=	config.context_length + max(config.lags_sequence  )

						__lowerCamelCase    					=	ids_tensor([self.batch_size, 1]							,			config.cardinality[0]  )
						__lowerCamelCase    					=	floats_tensor([self.batch_size, _past_length, config.num_time_features]  )
						__lowerCamelCase    					=	floats_tensor([self.batch_size, _past_length]  )
						__lowerCamelCase    					=	floats_tensor([self.batch_size, _past_length]  ) > 0.5

						# decoder inputs
						__lowerCamelCase    					=	floats_tensor([self.batch_size, config.prediction_length, config.num_time_features]  )
						__lowerCamelCase    					=	floats_tensor([self.batch_size, config.prediction_length]  )

						__lowerCamelCase    					=	{
						    '''past_values''': past_values,
						    '''static_categorical_features''': static_categorical_features,
						    '''past_time_features''': past_time_features,
						    '''past_observed_mask''': past_observed_mask,
						    '''future_time_features''': future_time_features,
						    '''future_values''': future_values,
						}
						return inputs_dict






		def  __A       (		self		:							int  )     ->							Optional[Any]:
						__lowerCamelCase    					=	self.get_config()
						__lowerCamelCase    					=	self.prepare_autoformer_inputs_dict(SCREAMING_SNAKE_CASE__  )
						return config, inputs_dict






		def  __A       (		self		:							Optional[int]  )     ->							Union[str, Any]:
						__lowerCamelCase					,    __lowerCamelCase    					=	self.prepare_config_and_inputs()
						return config, inputs_dict







		def  __A       (		self		:							Union[str, Any]							,			SCREAMING_SNAKE_CASE__		:							str							,			SCREAMING_SNAKE_CASE__		:							str  )     ->							Union[str, Any]:
						__lowerCamelCase    					=	AutoformerModel(config=SCREAMING_SNAKE_CASE__  ).to(SCREAMING_SNAKE_CASE__  ).eval()
						__lowerCamelCase    					=	model(**SCREAMING_SNAKE_CASE__  )

						__lowerCamelCase    					=	outputs.encoder_last_hidden_state
						__lowerCamelCase    					=	outputs.last_hidden_state

						with tempfile.TemporaryDirectory() as tmpdirname:
										__lowerCamelCase    					=	model.get_encoder()
										encoder.save_pretrained(SCREAMING_SNAKE_CASE__  )
										__lowerCamelCase    					=	AutoformerEncoder.from_pretrained(SCREAMING_SNAKE_CASE__  ).to(SCREAMING_SNAKE_CASE__  )

						__lowerCamelCase					,    __lowerCamelCase					,    __lowerCamelCase					,    __lowerCamelCase					,    __lowerCamelCase    					=	model.create_network_inputs(**SCREAMING_SNAKE_CASE__  )
						__lowerCamelCase					,    __lowerCamelCase    					=	model.decomposition_layer(transformer_inputs[:, : config.context_length, ...]  )

						__lowerCamelCase    					=	torch.cat(
						    (transformer_inputs[:, : config.context_length, ...], feature[:, : config.context_length, ...])							,			dim=-1							,			)
						__lowerCamelCase    					=	encoder(inputs_embeds=SCREAMING_SNAKE_CASE__  )[0]
						self.parent.assertTrue((encoder_last_hidden_state_a - encoder_last_hidden_state).abs().max().item() < 1e-3  )

						__lowerCamelCase    					=	(
						    torch.mean(transformer_inputs[:, : config.context_length, ...]							,			dim=1  )
						    .unsqueeze(1  )
						    .repeat(1							,			config.prediction_length							,			1  )
						)
						__lowerCamelCase    					=	torch.zeros(
						    [transformer_inputs.shape[0], config.prediction_length, transformer_inputs.shape[2]]							,			device=enc_input.device							,			)

						__lowerCamelCase    					=	torch.cat(
						    (
						        torch.cat((seasonal_input[:, -config.label_length :, ...], zeros)							,			dim=1  ),
						        feature[:, config.context_length - config.label_length :, ...],
						    )							,			dim=-1							,			)
						__lowerCamelCase    					=	torch.cat(
						    (
						        torch.cat((trend_input[:, -config.label_length :, ...], mean)							,			dim=1  ),
						        feature[:, config.context_length - config.label_length :, ...],
						    )							,			dim=-1							,			)

						with tempfile.TemporaryDirectory() as tmpdirname:
										__lowerCamelCase    					=	model.get_decoder()
										decoder.save_pretrained(SCREAMING_SNAKE_CASE__  )
										__lowerCamelCase    					=	AutoformerDecoder.from_pretrained(SCREAMING_SNAKE_CASE__  ).to(SCREAMING_SNAKE_CASE__  )

						__lowerCamelCase    					=	decoder(
						    trend=SCREAMING_SNAKE_CASE__							,			inputs_embeds=SCREAMING_SNAKE_CASE__							,			encoder_hidden_states=SCREAMING_SNAKE_CASE__							,			)[0]

						self.parent.assertTrue((last_hidden_state_a - last_hidden_state).abs().max().item() < 1e-3  )





@require_torch
class    lowerCAmelCase__     (	__lowercase  ,	__lowercase  ,	unittest.TestCase			):
		a__       :						str        =		(AutoformerModel, AutoformerForPrediction) if is_torch_available() else ()
		a__       :						int        =		(AutoformerForPrediction,) if is_torch_available() else ()
		a__       :						List[Any]        =		{"""feature-extraction""": AutoformerModel} if is_torch_available() else {}
		a__       :						int        =		False
		a__       :						Dict        =		False
		a__       :						Tuple        =		False
		a__       :						Tuple        =		False
		a__       :						str        =		False
		a__       :						List[Any]        =		False






		def  __A       (		self		:							Dict  )     ->							str:
						__lowerCamelCase    					=	AutoformerModelTester(self  )
						__lowerCamelCase    					=	ConfigTester(self							,			config_class=SCREAMING_SNAKE_CASE__							,			has_text_modality=SCREAMING_SNAKE_CASE__  )






		def  __A       (		self		:							Any  )     ->							List[str]:
						self.config_tester.run_common_tests()






		def  __A       (		self		:							List[Any]  )     ->							Optional[Any]:
						__lowerCamelCase					,    __lowerCamelCase    					=	self.model_tester.prepare_config_and_inputs()
						for model_class in self.all_model_classes:
										__lowerCamelCase    					=	model_class(SCREAMING_SNAKE_CASE__  )

										with tempfile.TemporaryDirectory() as tmpdirname:
														model.save_pretrained(SCREAMING_SNAKE_CASE__  )
														__lowerCamelCase					,    __lowerCamelCase    					=	model_class.from_pretrained(SCREAMING_SNAKE_CASE__							,			output_loading_info=SCREAMING_SNAKE_CASE__  )
										self.assertEqual(info['''missing_keys''']							,			[]  )






		def  __A       (		self		:							Optional[Any]  )     ->							Union[str, Any]:
						__lowerCamelCase    					=	self.model_tester.prepare_config_and_inputs_for_common()
						self.model_tester.check_encoder_decoder_model_standalone(*SCREAMING_SNAKE_CASE__  )






		@unittest.skip(reason='''Model has no tokens embeddings'''  )
		def  __A       (		self		:							List[Any]  )     ->							Union[str, Any]:
						pass






		def  __A       (		self		:							int  )     ->							Tuple:
						__lowerCamelCase    					=	inspect.signature(getattr(SCREAMING_SNAKE_CASE__							,			'''forward'''  )  )
						# The main input is the name of the argument after `self`
						__lowerCamelCase    					=	list(model_signature.parameters.keys()  )[1]
						self.assertEqual(AutoformerModel.main_input_name							,			SCREAMING_SNAKE_CASE__  )






		def  __A       (		self		:							Optional[int]  )     ->							List[Any]:
						__lowerCamelCase					,    __lowerCamelCase    					=	self.model_tester.prepare_config_and_inputs_for_common()

						for model_class in self.all_model_classes:
										__lowerCamelCase    					=	model_class(SCREAMING_SNAKE_CASE__  )
										__lowerCamelCase    					=	inspect.signature(model.forward  )
										# signature.parameters is an OrderedDict => so arg_names order is deterministic
										__lowerCamelCase    					=	[*signature.parameters.keys()]

										__lowerCamelCase    					=	[
										    '''past_values''',
										    '''past_time_features''',
										    '''past_observed_mask''',
										    '''static_categorical_features''',
										    '''static_real_features''',
										    '''future_values''',
										    '''future_time_features''',
										]

										if model.__class__.__name__ in ["AutoformerForPrediction"]:
														expected_arg_names.append('''future_observed_mask'''  )

										expected_arg_names.extend(
										    [
										        '''decoder_attention_mask''',
										        '''head_mask''',
										        '''decoder_head_mask''',
										        '''cross_attn_head_mask''',
										        '''encoder_outputs''',
										        '''past_key_values''',
										        '''output_hidden_states''',
										        '''output_attentions''',
										        '''use_cache''',
										        '''return_dict''',
										    ]  )

										self.assertListEqual(arg_names[: len(SCREAMING_SNAKE_CASE__  )]							,			SCREAMING_SNAKE_CASE__  )






		def  __A       (		self		:							List[Any]  )     ->							Optional[Any]:
						__lowerCamelCase					,    __lowerCamelCase    					=	self.model_tester.prepare_config_and_inputs_for_common()
						__lowerCamelCase    					=	True

						__lowerCamelCase    					=	getattr(self.model_tester							,			'''seq_length'''							,			SCREAMING_SNAKE_CASE__  )
						__lowerCamelCase    					=	getattr(self.model_tester							,			'''decoder_seq_length'''							,			SCREAMING_SNAKE_CASE__  )
						__lowerCamelCase    					=	getattr(self.model_tester							,			'''encoder_seq_length'''							,			SCREAMING_SNAKE_CASE__  )
						__lowerCamelCase    					=	getattr(self.model_tester							,			'''d_model'''							,			SCREAMING_SNAKE_CASE__  )
						__lowerCamelCase    					=	getattr(self.model_tester							,			'''num_attention_heads'''							,			SCREAMING_SNAKE_CASE__  )
						__lowerCamelCase    					=	d_model // num_attention_heads

						for model_class in self.all_model_classes:
										__lowerCamelCase    					=	True
										__lowerCamelCase    					=	False
										__lowerCamelCase    					=	True
										__lowerCamelCase    					=	model_class(SCREAMING_SNAKE_CASE__  )
										model.to(SCREAMING_SNAKE_CASE__  )
										model.eval()
										with torch.no_grad():
														__lowerCamelCase    					=	model(**self._prepare_for_class(SCREAMING_SNAKE_CASE__							,			SCREAMING_SNAKE_CASE__  )  )
										__lowerCamelCase    					=	outputs.encoder_attentions if config.is_encoder_decoder else outputs.attentions
										self.assertEqual(len(SCREAMING_SNAKE_CASE__  )							,			self.model_tester.num_hidden_layers  )

										# check that output_attentions also work using config
										del inputs_dict["output_attentions"]
										__lowerCamelCase    					=	True
										__lowerCamelCase    					=	model_class(SCREAMING_SNAKE_CASE__  )
										model.to(SCREAMING_SNAKE_CASE__  )
										model.eval()
										with torch.no_grad():
														__lowerCamelCase    					=	model(**self._prepare_for_class(SCREAMING_SNAKE_CASE__							,			SCREAMING_SNAKE_CASE__  )  )
										__lowerCamelCase    					=	outputs.encoder_attentions
										self.assertEqual(len(SCREAMING_SNAKE_CASE__  )							,			self.model_tester.num_hidden_layers  )

										self.assertListEqual(
										    list(attentions[0].shape[-3:]  )							,			[self.model_tester.num_attention_heads, encoder_seq_length, dim]							,			)
										__lowerCamelCase    					=	len(SCREAMING_SNAKE_CASE__  )

										__lowerCamelCase    					=	7

										if "last_hidden_state" in outputs:
														correct_outlen += 1

										if "trend" in outputs:
														correct_outlen += 1

										if "past_key_values" in outputs:
														correct_outlen += 1  # past_key_values have been returned

										if "loss" in outputs:
														correct_outlen += 1

										if "params" in outputs:
														correct_outlen += 1

										self.assertEqual(SCREAMING_SNAKE_CASE__							,			SCREAMING_SNAKE_CASE__  )

										# decoder attentions
										__lowerCamelCase    					=	outputs.decoder_attentions
										self.assertIsInstance(SCREAMING_SNAKE_CASE__							,			(list, tuple)  )
										self.assertEqual(len(SCREAMING_SNAKE_CASE__  )							,			self.model_tester.num_hidden_layers  )
										self.assertListEqual(
										    list(decoder_attentions[0].shape[-3:]  )							,			[self.model_tester.num_attention_heads, decoder_seq_length, dim]							,			)

										# cross attentions
										__lowerCamelCase    					=	outputs.cross_attentions
										self.assertIsInstance(SCREAMING_SNAKE_CASE__							,			(list, tuple)  )
										self.assertEqual(len(SCREAMING_SNAKE_CASE__  )							,			self.model_tester.num_hidden_layers  )
										self.assertListEqual(
										    list(cross_attentions[0].shape[-3:]  )							,			[self.model_tester.num_attention_heads, decoder_seq_length, dim]							,			)

						# Check attention is always last and order is fine
						__lowerCamelCase    					=	True
						__lowerCamelCase    					=	True
						__lowerCamelCase    					=	model_class(SCREAMING_SNAKE_CASE__  )
						model.to(SCREAMING_SNAKE_CASE__  )
						model.eval()
						with torch.no_grad():
										__lowerCamelCase    					=	model(**self._prepare_for_class(SCREAMING_SNAKE_CASE__							,			SCREAMING_SNAKE_CASE__  )  )

						self.assertEqual(out_len + 2							,			len(SCREAMING_SNAKE_CASE__  )  )

						__lowerCamelCase    					=	outputs.encoder_attentions if config.is_encoder_decoder else outputs.attentions

						self.assertEqual(len(SCREAMING_SNAKE_CASE__  )							,			self.model_tester.num_hidden_layers  )
						self.assertListEqual(
						    list(self_attentions[0].shape[-3:]  )							,			[self.model_tester.num_attention_heads, encoder_seq_length, dim]							,			)







		@is_flaky()
		def  __A       (		self		:							List[str]  )     ->							Optional[int]:
						super().test_retain_grad_hidden_states_attentions()





def  __magic_name__ (   __lowerCAmelCase	:						Optional[int]="train-batch.pt"							)			->   Union[str, Any]:
				__lowerCamelCase    					=	hf_hub_download(repo_id='''hf-internal-testing/tourism-monthly-batch'''		,  filename=__lowerCAmelCase		,  repo_type='''dataset'''							)
				__lowerCamelCase    					=	torch.load(__lowerCAmelCase		,  map_location=__lowerCAmelCase							)
				return batch







@require_torch
@slow
class    lowerCAmelCase__     (	unittest.TestCase			):






		def  __A       (		self		:							Tuple  )     ->							Union[str, Any]:
						__lowerCamelCase    					=	AutoformerModel.from_pretrained('''huggingface/autoformer-tourism-monthly'''  ).to(SCREAMING_SNAKE_CASE__  )
						__lowerCamelCase    					=	prepare_batch()

						with torch.no_grad():
										__lowerCamelCase    					=	model(
										    past_values=batch['''past_values''']							,			past_time_features=batch['''past_time_features''']							,			past_observed_mask=batch['''past_observed_mask''']							,			static_categorical_features=batch['''static_categorical_features''']							,			future_values=batch['''future_values''']							,			future_time_features=batch['''future_time_features''']							,			)[0]

						__lowerCamelCase    					=	torch.Size(
						    (64, model.config.prediction_length + model.config.label_length, model.config.feature_size)  )
						self.assertEqual(output.shape							,			SCREAMING_SNAKE_CASE__  )

						__lowerCamelCase    					=	torch.tensor(
						    [[0.3593, -1.3398, 0.6330], [0.2279, 1.5396, -0.1792], [0.0450, 1.3225, -0.2335]]							,			device=SCREAMING_SNAKE_CASE__  )
						self.assertTrue(torch.allclose(output[0, :3, :3]							,			SCREAMING_SNAKE_CASE__							,			atol=SCREAMING_SNAKE_CASE__  )  )






		def  __A       (		self		:							Dict  )     ->							Optional[int]:
						__lowerCamelCase    					=	AutoformerForPrediction.from_pretrained('''huggingface/autoformer-tourism-monthly'''  ).to(SCREAMING_SNAKE_CASE__  )
						__lowerCamelCase    					=	prepare_batch('''val-batch.pt'''  )
						with torch.no_grad():
										__lowerCamelCase    					=	model(
										    past_values=batch['''past_values''']							,			past_time_features=batch['''past_time_features''']							,			past_observed_mask=batch['''past_observed_mask''']							,			static_categorical_features=batch['''static_categorical_features''']							,			).encoder_last_hidden_state
						__lowerCamelCase    					=	torch.Size((64, model.config.context_length, model.config.d_model)  )
						self.assertEqual(output.shape							,			SCREAMING_SNAKE_CASE__  )

						__lowerCamelCase    					=	torch.tensor(
						    [[-0.0734, -0.9036, 0.8358], [4.7186, 2.4113, 1.9581], [1.7953, 2.3558, 1.2970]]							,			device=SCREAMING_SNAKE_CASE__  )
						self.assertTrue(torch.allclose(output[0, :3, :3]							,			SCREAMING_SNAKE_CASE__							,			atol=SCREAMING_SNAKE_CASE__  )  )







		def  __A       (		self		:							int  )     ->							List[Any]:
						__lowerCamelCase    					=	AutoformerForPrediction.from_pretrained('''huggingface/autoformer-tourism-monthly'''  ).to(SCREAMING_SNAKE_CASE__  )
						__lowerCamelCase    					=	prepare_batch('''val-batch.pt'''  )
						with torch.no_grad():
										__lowerCamelCase    					=	model.generate(
										    static_categorical_features=batch['''static_categorical_features''']							,			past_time_features=batch['''past_time_features''']							,			past_values=batch['''past_values''']							,			future_time_features=batch['''future_time_features''']							,			past_observed_mask=batch['''past_observed_mask''']							,			)
						__lowerCamelCase    					=	torch.Size((64, model.config.num_parallel_samples, model.config.prediction_length)  )
						self.assertEqual(outputs.sequences.shape							,			SCREAMING_SNAKE_CASE__  )

						__lowerCamelCase    					=	torch.tensor([3130.6763, 4056.5293, 7053.0786]							,			device=SCREAMING_SNAKE_CASE__  )
						__lowerCamelCase    					=	outputs.sequences.mean(dim=1  )
						self.assertTrue(torch.allclose(mean_prediction[0, -3:]							,			SCREAMING_SNAKE_CASE__							,			rtol=1e-1  )  )





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""": 1_600, """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""": 1_600, """eval_accuracy""": 0.3, """eval_loss""": 1.2},
        },
    ]			)
class    lowerCAmelCase__     (	unittest.TestCase			):






		def  __A       (		self		:							Optional[int]  )     ->							List[Any]:
						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=SCREAMING_SNAKE_CASE__							,			)
						assert hasattr(self							,			'''env'''  )






		def  __A       (		self		:							List[str]							,			SCREAMING_SNAKE_CASE__		:							List[Any]  )     ->							Optional[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=SCREAMING_SNAKE_CASE__							,			instance_type=self.instance_type							,			debugger_hook_config=SCREAMING_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=SCREAMING_SNAKE_CASE__							,			py_version='''py36'''							,			)






		def  __A       (		self		:							List[Any]							,			SCREAMING_SNAKE_CASE__		:							List[Any]  )     ->							List[str]:
						TrainingJobAnalytics(SCREAMING_SNAKE_CASE__  ).export_csv(f'''{self.env.test_path}/{job_name}_metrics.csv'''  )







		@parameterized.expand([(1,)]  )
		def  __A       (		self		:							Optional[Any]							,			SCREAMING_SNAKE_CASE__		:							Optional[Any]  )     ->							Optional[int]:
						# create estimator
						__lowerCamelCase    					=	self.create_estimator(SCREAMING_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}							,			SCREAMING_SNAKE_CASE__  )



