



"""simple docstring"""




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


lowerCAmelCase__		:					int							=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        UpperCAmelCase_  :

						'''simple docstring'''




						def __init__(   self				,							_A				,							_A=16				,							_A=13				,							_A=7				,							_A=14				,							_A=10				,							_A=19				,							_A=5				,							_A=4				,							_A=True				,							_A=16				,							_A=2				,							_A=4				,							_A=4				,							_A="gelu"				,							_A=0.1				,							_A=0.1				,							_A=[1, 2, 3, 4, 5]				,							_A=25				,							_A=5				,							):







													'''simple docstring'''

													__SCREAMING_SNAKE_CASE									= d_model
													__SCREAMING_SNAKE_CASE									= parent
													__SCREAMING_SNAKE_CASE									= batch_size
													__SCREAMING_SNAKE_CASE									= prediction_length
													__SCREAMING_SNAKE_CASE									= context_length
													__SCREAMING_SNAKE_CASE									= cardinality
													__SCREAMING_SNAKE_CASE									= num_time_features
													__SCREAMING_SNAKE_CASE									= lags_sequence
													__SCREAMING_SNAKE_CASE									= embedding_dimension
													__SCREAMING_SNAKE_CASE									= is_training
													__SCREAMING_SNAKE_CASE									= hidden_size
													__SCREAMING_SNAKE_CASE									= num_hidden_layers
													__SCREAMING_SNAKE_CASE									= num_attention_heads
													__SCREAMING_SNAKE_CASE									= intermediate_size
													__SCREAMING_SNAKE_CASE									= hidden_act
													__SCREAMING_SNAKE_CASE									= hidden_dropout_prob
													__SCREAMING_SNAKE_CASE									= attention_probs_dropout_prob

													__SCREAMING_SNAKE_CASE									= context_length
													__SCREAMING_SNAKE_CASE									= prediction_length + label_length
													__SCREAMING_SNAKE_CASE									= label_length

													__SCREAMING_SNAKE_CASE									= moving_average
													__SCREAMING_SNAKE_CASE									= autocorrelation_factor






						def 	_A     (   self		):







													'''simple docstring'''

													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				,							_A		):







													'''simple docstring'''

													__SCREAMING_SNAKE_CASE									= config.context_length + max(config.lags_sequence		)

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

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

													__SCREAMING_SNAKE_CASE									= {
													    '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		):







													'''simple docstring'''

													__SCREAMING_SNAKE_CASE									= self.get_config()
													__SCREAMING_SNAKE_CASE									= self.prepare_autoformer_inputs_dict(__A		)
													return config, inputs_dict






						def 	_A     (   self		):







													'''simple docstring'''

													__SCREAMING_SNAKE_CASE							,	__SCREAMING_SNAKE_CASE									= self.prepare_config_and_inputs()
													return config, inputs_dict






						def 	_A     (   self				,							_A				,							_A		):







													'''simple docstring'''

													__SCREAMING_SNAKE_CASE									= AutoformerModel(config=__A		).to(__A		).eval()
													__SCREAMING_SNAKE_CASE									= model(**__A		)

													__SCREAMING_SNAKE_CASE									= outputs.encoder_last_hidden_state
													__SCREAMING_SNAKE_CASE									= outputs.last_hidden_state

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

													__SCREAMING_SNAKE_CASE							,	__SCREAMING_SNAKE_CASE							,	__SCREAMING_SNAKE_CASE							,	__SCREAMING_SNAKE_CASE							,	__SCREAMING_SNAKE_CASE									= model.create_network_inputs(**__A		)
													__SCREAMING_SNAKE_CASE							,	__SCREAMING_SNAKE_CASE									= model.decomposition_layer(transformer_inputs[:, : config.context_length, ...]		)

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

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

													__SCREAMING_SNAKE_CASE									= torch.cat(
													    (
													        torch.cat((seasonal_input[:, -config.label_length :, ...], zeros)				,							dim=1		),
													        feature[:, config.context_length - config.label_length :, ...],
													    )				,							dim=-1				,							)
													__SCREAMING_SNAKE_CASE									= 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:
																				__SCREAMING_SNAKE_CASE									= model.get_decoder()
																				decoder.save_pretrained(__A		)
																				__SCREAMING_SNAKE_CASE									= AutoformerDecoder.from_pretrained(__A		).to(__A		)

													__SCREAMING_SNAKE_CASE									= decoder(
													    trend=__A				,							inputs_embeds=__A				,							encoder_hidden_states=__A				,							)[0]

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







@require_torch
class        UpperCAmelCase_  (       UpperCamelCase_   ,   UpperCamelCase_   ,   unittest.TestCase	):

						'''simple docstring'''




						UpperCamelCase__      :			Optional[int]      =  (AutoformerModel, AutoformerForPrediction) if is_torch_available() else ()
						UpperCamelCase__      :			Union[str, Any]      =  (AutoformerForPrediction,) if is_torch_available() else ()
						UpperCamelCase__      :			Tuple      =  {'''feature-extraction''': AutoformerModel} if is_torch_available() else {}
						UpperCamelCase__      :			List[str]      =  False
						UpperCamelCase__      :			Union[str, Any]      =  False
						UpperCamelCase__      :			int      =  False
						UpperCamelCase__      :			int      =  False
						UpperCamelCase__      :			List[Any]      =  False
						UpperCamelCase__      :			str      =  False






						def 	_A     (   self		):







													'''simple docstring'''

													__SCREAMING_SNAKE_CASE									= AutoformerModelTester(self		)
													__SCREAMING_SNAKE_CASE									= ConfigTester(self				,							config_class=__A				,							has_text_modality=__A		)






						def 	_A     (   self		):







													'''simple docstring'''

													self.config_tester.run_common_tests()






						def 	_A     (   self		):







													'''simple docstring'''

													__SCREAMING_SNAKE_CASE							,	__SCREAMING_SNAKE_CASE									= self.model_tester.prepare_config_and_inputs()
													for model_class in self.all_model_classes:
																				__SCREAMING_SNAKE_CASE									= model_class(__A		)

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






						def 	_A     (   self		):







													'''simple docstring'''

													__SCREAMING_SNAKE_CASE									= self.model_tester.prepare_config_and_inputs_for_common()
													self.model_tester.check_encoder_decoder_model_standalone(*__A		)






						@unittest.skip(reason='Model has no tokens embeddings'		)
						def 	_A     (   self		):







													'''simple docstring'''

													pass






						def 	_A     (   self		):







													'''simple docstring'''

													__SCREAMING_SNAKE_CASE									= inspect.signature(getattr(__A				,							'forward'		)		)
													# The main input is the name of the argument after `self`
													__SCREAMING_SNAKE_CASE									= list(model_signature.parameters.keys()		)[1]
													self.assertEqual(AutoformerModel.main_input_name				,							__A		)






						def 	_A     (   self		):







													'''simple docstring'''

													__SCREAMING_SNAKE_CASE							,	__SCREAMING_SNAKE_CASE									= self.model_tester.prepare_config_and_inputs_for_common()

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

																				__SCREAMING_SNAKE_CASE									= [
																				    '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(__A		)]				,							__A		)






						def 	_A     (   self		):







													'''simple docstring'''

													__SCREAMING_SNAKE_CASE							,	__SCREAMING_SNAKE_CASE									= self.model_tester.prepare_config_and_inputs_for_common()
													__SCREAMING_SNAKE_CASE									= True

													__SCREAMING_SNAKE_CASE									= getattr(self.model_tester				,							'seq_length'				,							__A		)
													__SCREAMING_SNAKE_CASE									= getattr(self.model_tester				,							'decoder_seq_length'				,							__A		)
													__SCREAMING_SNAKE_CASE									= getattr(self.model_tester				,							'encoder_seq_length'				,							__A		)
													__SCREAMING_SNAKE_CASE									= getattr(self.model_tester				,							'd_model'				,							__A		)
													__SCREAMING_SNAKE_CASE									= getattr(self.model_tester				,							'num_attention_heads'				,							__A		)
													__SCREAMING_SNAKE_CASE									= d_model // num_attention_heads

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

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

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

																				__SCREAMING_SNAKE_CASE									= 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(__A				,							__A		)

																				# decoder attentions
																				__SCREAMING_SNAKE_CASE									= outputs.decoder_attentions
																				self.assertIsInstance(__A				,							(list, tuple)		)
																				self.assertEqual(len(__A		)				,							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
																				__SCREAMING_SNAKE_CASE									= outputs.cross_attentions
																				self.assertIsInstance(__A				,							(list, tuple)		)
																				self.assertEqual(len(__A		)				,							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
													__SCREAMING_SNAKE_CASE									= True
													__SCREAMING_SNAKE_CASE									= True
													__SCREAMING_SNAKE_CASE									= model_class(__A		)
													model.to(__A		)
													model.eval()
													with torch.no_grad():
																				__SCREAMING_SNAKE_CASE									= model(**self._prepare_for_class(__A				,							__A		)		)

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

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

													self.assertEqual(len(__A		)				,							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		):







													'''simple docstring'''

													super().test_retain_grad_hidden_states_attentions()







def      __lowercase       (			a__="train-batch.pt"       )    ->			Optional[int]:
							__SCREAMING_SNAKE_CASE									= hf_hub_download(repo_id='hf-internal-testing/tourism-monthly-batch'							,       filename=_lowercase							,       repo_type='dataset'       )
							__SCREAMING_SNAKE_CASE									= torch.load(_lowercase							,       map_location=_lowercase       )
							return batch

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

						'''simple docstring'''




						def 	_A     (   self		):







													'''simple docstring'''

													__SCREAMING_SNAKE_CASE									= AutoformerModel.from_pretrained('huggingface/autoformer-tourism-monthly'		).to(__A		)
													__SCREAMING_SNAKE_CASE									= prepare_batch()

													with torch.no_grad():
																				__SCREAMING_SNAKE_CASE									= 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]

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

													__SCREAMING_SNAKE_CASE									= torch.tensor(
													    [[0.3_5_9_3, -1.3_3_9_8, 0.6_3_3_0], [0.2_2_7_9, 1.5_3_9_6, -0.1_7_9_2], [0.0_4_5_0, 1.3_2_2_5, -0.2_3_3_5]]				,							device=__A		)
													self.assertTrue(torch.allclose(output[0, :3, :3]				,							__A				,							atol=__A		)		)






						def 	_A     (   self		):







													'''simple docstring'''

													__SCREAMING_SNAKE_CASE									= AutoformerForPrediction.from_pretrained('huggingface/autoformer-tourism-monthly'		).to(__A		)
													__SCREAMING_SNAKE_CASE									= prepare_batch('val-batch.pt'		)
													with torch.no_grad():
																				__SCREAMING_SNAKE_CASE									= 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
													__SCREAMING_SNAKE_CASE									= torch.Size((64, model.config.context_length, model.config.d_model)		)
													self.assertEqual(output.shape				,							__A		)

													__SCREAMING_SNAKE_CASE									= torch.tensor(
													    [[-0.0_7_3_4, -0.9_0_3_6, 0.8_3_5_8], [4.7_1_8_6, 2.4_1_1_3, 1.9_5_8_1], [1.7_9_5_3, 2.3_5_5_8, 1.2_9_7_0]]				,							device=__A		)
													self.assertTrue(torch.allclose(output[0, :3, :3]				,							__A				,							atol=__A		)		)






						def 	_A     (   self		):







													'''simple docstring'''

													__SCREAMING_SNAKE_CASE									= AutoformerForPrediction.from_pretrained('huggingface/autoformer-tourism-monthly'		).to(__A		)
													__SCREAMING_SNAKE_CASE									= prepare_batch('val-batch.pt'		)
													with torch.no_grad():
																				__SCREAMING_SNAKE_CASE									= 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']				,							)
													__SCREAMING_SNAKE_CASE									= torch.Size((64, model.config.num_parallel_samples, model.config.prediction_length)		)
													self.assertEqual(outputs.sequences.shape				,							__A		)

													__SCREAMING_SNAKE_CASE									= torch.tensor([3_130.6_763, 4_056.5_293, 7_053.0_786]				,							device=__A		)
													__SCREAMING_SNAKE_CASE									= outputs.sequences.mean(dim=1		)
													self.assertTrue(torch.allclose(mean_prediction[0, -3:]				,							__A				,							rtol=1e-1		)		)











from ..utils import DummyObject, requires_backends

class        UpperCAmelCase_  (       metaclass=UpperCamelCase_	):

						'''simple docstring'''




						UpperCamelCase__      :			Optional[int]      =  ['''speech''']






						def __init__(   self				,							*_A				,							**_A		):







													'''simple docstring'''

													requires_backends(self				,							['speech']		)







class        UpperCAmelCase_  (       metaclass=UpperCamelCase_	):

						'''simple docstring'''




						UpperCamelCase__      :			Optional[int]      =  ['''speech''']






						def __init__(   self				,							*_A				,							**_A		):







													'''simple docstring'''

													requires_backends(self				,							['speech']		)






