
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    :      Union[str, Any]          =			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				,       UpperCamelCase__				,       UpperCamelCase__=16				,       UpperCamelCase__=13				,       UpperCamelCase__=7				,       UpperCamelCase__=14				,       UpperCamelCase__=10				,       UpperCamelCase__=19				,       UpperCamelCase__=5				,       UpperCamelCase__=4				,       UpperCamelCase__=True				,       UpperCamelCase__=16				,       UpperCamelCase__=2				,       UpperCamelCase__=4				,       UpperCamelCase__=4				,       UpperCamelCase__="gelu"				,       UpperCamelCase__=0.1				,       UpperCamelCase__=0.1				,       UpperCamelCase__=[1, 2, 3, 4, 5]				,       UpperCamelCase__=25				,       UpperCamelCase__=5				,       ):
					A__  :	Optional[int]         =							d_model
					A__  :	Dict         =							parent
					A__  :	List[Any]         =							batch_size
					A__  :	Dict         =							prediction_length
					A__  :	List[Any]         =							context_length
					A__  :	Union[str, Any]         =							cardinality
					A__  :	int         =							num_time_features
					A__  :	Optional[Any]         =							lags_sequence
					A__  :	Optional[Any]         =							embedding_dimension
					A__  :	Any         =							is_training
					A__  :	Dict         =							hidden_size
					A__  :	List[Any]         =							num_hidden_layers
					A__  :	List[str]         =							num_attention_heads
					A__  :	Optional[int]         =							intermediate_size
					A__  :	Optional[Any]         =							hidden_act
					A__  :	Optional[int]         =							hidden_dropout_prob
					A__  :	Any         =							attention_probs_dropout_prob

					A__  :	List[str]         =							context_length
					A__  :	Any         =							prediction_length + label_length
					A__  :	List[str]         =							label_length

					A__  :	str         =							moving_average
					A__  :	List[str]         =							autocorrelation_factor





	def   __snake_case				(			self		):
					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   __snake_case				(			self				,       UpperCamelCase__		):
					A__  :	int         =							config.context_length + max(config.lags_sequence		)

					A__  :	Any         =							ids_tensor([self.batch_size, 1]				,       config.cardinality[0]		)
					A__  :	List[str]         =							floats_tensor([self.batch_size, _past_length, config.num_time_features]		)
					A__  :	int         =							floats_tensor([self.batch_size, _past_length]		)
					A__  :	Optional[int]         =							floats_tensor([self.batch_size, _past_length]		) > 0.5

					# decoder inputs
					A__  :	Optional[Any]         =							floats_tensor([self.batch_size, config.prediction_length, config.num_time_features]		)
					A__  :	Optional[int]         =							floats_tensor([self.batch_size, config.prediction_length]		)

					A__  :	str         =							{
					    '''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   __snake_case				(			self		):
					A__  :	Tuple         =							self.get_config()
					A__  :	Tuple         =							self.prepare_autoformer_inputs_dict(UpperCamelCase__		)
					return config, inputs_dict





	def   __snake_case				(			self		):
					A__  :	Tuple         =							self.prepare_config_and_inputs()
					return config, inputs_dict





	def   __snake_case				(			self				,       UpperCamelCase__				,       UpperCamelCase__		):
					A__  :	Optional[int]         =							AutoformerModel(config=UpperCamelCase__		).to(UpperCamelCase__		).eval()
					A__  :	List[str]         =							model(**UpperCamelCase__		)

					A__  :	Any         =							outputs.encoder_last_hidden_state
					A__  :	str         =							outputs.last_hidden_state

					with tempfile.TemporaryDirectory() as tmpdirname:
									A__  :	Tuple         =							model.get_encoder()
									encoder.save_pretrained(UpperCamelCase__		)
									A__  :	Union[str, Any]         =							AutoformerEncoder.from_pretrained(UpperCamelCase__		).to(UpperCamelCase__		)

					A__  :	str         =							model.create_network_inputs(**UpperCamelCase__		)
					A__  :	Dict         =							model.decomposition_layer(transformer_inputs[:, : config.context_length, ...]		)

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

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

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

					A__  :	Any         =							decoder(
					    trend=UpperCamelCase__				,       inputs_embeds=UpperCamelCase__				,       encoder_hidden_states=UpperCamelCase__				,       )[0]

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







@require_torch
class  UpperCamelCase__		(    SCREAMING_SNAKE_CASE_,			SCREAMING_SNAKE_CASE_,			unittest.TestCase   ):







	'''simple docstring'''
	_lowerCAmelCase							       =				(AutoformerModel, AutoformerForPrediction) if is_torch_available() else ()
	_lowerCAmelCase							       =				(AutoformerForPrediction,) if is_torch_available() else ()
	_lowerCAmelCase							       =				{"feature-extraction": AutoformerModel} if is_torch_available() else {}
	_lowerCAmelCase							       =				False
	_lowerCAmelCase							       =				False
	_lowerCAmelCase							       =				False
	_lowerCAmelCase							       =				False
	_lowerCAmelCase							       =				False
	_lowerCAmelCase							       =				False





	def   __snake_case				(			self		):
					A__  :	List[str]         =							AutoformerModelTester(self		)
					A__  :	Dict         =							ConfigTester(self				,       config_class=UpperCamelCase__				,       has_text_modality=UpperCamelCase__		)





	def   __snake_case				(			self		):
					self.config_tester.run_common_tests()





	def   __snake_case				(			self		):
					A__  :	Dict         =							self.model_tester.prepare_config_and_inputs()
					for model_class in self.all_model_classes:
									A__  :	int         =							model_class(UpperCamelCase__		)

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





	def   __snake_case				(			self		):
					A__  :	List[Any]         =							self.model_tester.prepare_config_and_inputs_for_common()
					self.model_tester.check_encoder_decoder_model_standalone(*UpperCamelCase__		)





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





	def   __snake_case				(			self		):
					A__  :	Optional[Any]         =							inspect.signature(getattr(UpperCamelCase__				,       '''forward'''		)		)
					# The main input is the name of the argument after `self`
					A__  :	Tuple         =							list(model_signature.parameters.keys()		)[1]
					self.assertEqual(AutoformerModel.main_input_name				,       UpperCamelCase__		)





	def   __snake_case				(			self		):
					A__  :	Any         =							self.model_tester.prepare_config_and_inputs_for_common()

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

									A__  :	int         =							[
									    '''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(UpperCamelCase__		)]				,       UpperCamelCase__		)





	def   __snake_case				(			self		):
					A__  :	Dict         =							self.model_tester.prepare_config_and_inputs_for_common()
					A__  :	Dict         =							True

					A__  :	int         =							getattr(self.model_tester				,       '''seq_length'''				,       UpperCamelCase__		)
					A__  :	Union[str, Any]         =							getattr(self.model_tester				,       '''decoder_seq_length'''				,       UpperCamelCase__		)
					A__  :	Tuple         =							getattr(self.model_tester				,       '''encoder_seq_length'''				,       UpperCamelCase__		)
					A__  :	Dict         =							getattr(self.model_tester				,       '''d_model'''				,       UpperCamelCase__		)
					A__  :	Dict         =							getattr(self.model_tester				,       '''num_attention_heads'''				,       UpperCamelCase__		)
					A__  :	Union[str, Any]         =							d_model // num_attention_heads

					for model_class in self.all_model_classes:
									A__  :	List[Any]         =							True
									A__  :	Optional[int]         =							False
									A__  :	Union[str, Any]         =							True
									A__  :	Any         =							model_class(UpperCamelCase__		)
									model.to(UpperCamelCase__		)
									model.eval()
									with torch.no_grad():
													A__  :	List[str]         =							model(**self._prepare_for_class(UpperCamelCase__				,       UpperCamelCase__		)		)
									A__  :	List[Any]         =							outputs.encoder_attentions if config.is_encoder_decoder else outputs.attentions
									self.assertEqual(len(UpperCamelCase__		)				,       self.model_tester.num_hidden_layers		)

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

									self.assertListEqual(
									    list(attentions[0].shape[-3:]		)				,       [self.model_tester.num_attention_heads, encoder_seq_length, dim]				,       )
									A__  :	Union[str, Any]         =							len(UpperCamelCase__		)

									A__  :	List[Any]         =							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(UpperCamelCase__				,       UpperCamelCase__		)

									# decoder attentions
									A__  :	str         =							outputs.decoder_attentions
									self.assertIsInstance(UpperCamelCase__				,       (list, tuple)		)
									self.assertEqual(len(UpperCamelCase__		)				,       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
									A__  :	Optional[Any]         =							outputs.cross_attentions
									self.assertIsInstance(UpperCamelCase__				,       (list, tuple)		)
									self.assertEqual(len(UpperCamelCase__		)				,       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
					A__  :	Optional[int]         =							True
					A__  :	Optional[int]         =							True
					A__  :	Tuple         =							model_class(UpperCamelCase__		)
					model.to(UpperCamelCase__		)
					model.eval()
					with torch.no_grad():
									A__  :	Dict         =							model(**self._prepare_for_class(UpperCamelCase__				,       UpperCamelCase__		)		)

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

					A__  :	Optional[int]         =							outputs.encoder_attentions if config.is_encoder_decoder else outputs.attentions

					self.assertEqual(len(UpperCamelCase__		)				,       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   __snake_case				(			self		):
					super().test_retain_grad_hidden_states_attentions()




def 				SCREAMING_SNAKE_CASE     (      __UpperCamelCase :						Union[str, Any]="train-batch.pt"       ) ->       str:

				"""simple docstring"""







				A__  :	Dict         =							hf_hub_download(repo_id='''hf-internal-testing/tourism-monthly-batch'''							,						filename=__UpperCamelCase							,						repo_type='''dataset'''       )
				A__  :	Tuple         =							torch.load(__UpperCamelCase							,						map_location=__UpperCamelCase       )
				return batch







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







	'''simple docstring'''





	def   __snake_case				(			self		):
					A__  :	Dict         =							AutoformerModel.from_pretrained('''huggingface/autoformer-tourism-monthly'''		).to(UpperCamelCase__		)
					A__  :	int         =							prepare_batch()

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

					A__  :	Any         =							torch.Size(
					    (64, model.config.prediction_length + model.config.label_length, model.config.feature_size)		)
					self.assertEqual(output.shape				,       UpperCamelCase__		)

					A__  :	Any         =							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=UpperCamelCase__		)
					self.assertTrue(torch.allclose(output[0, :3, :3]				,       UpperCamelCase__				,       atol=UpperCamelCase__		)		)





	def   __snake_case				(			self		):
					A__  :	Tuple         =							AutoformerForPrediction.from_pretrained('''huggingface/autoformer-tourism-monthly'''		).to(UpperCamelCase__		)
					A__  :	Optional[Any]         =							prepare_batch('''val-batch.pt'''		)
					with torch.no_grad():
									A__  :	int         =							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
					A__  :	Tuple         =							torch.Size((64, model.config.context_length, model.config.d_model)		)
					self.assertEqual(output.shape				,       UpperCamelCase__		)

					A__  :	Dict         =							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=UpperCamelCase__		)
					self.assertTrue(torch.allclose(output[0, :3, :3]				,       UpperCamelCase__				,       atol=UpperCamelCase__		)		)





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

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

from numpy import exp, pi, sqrt

def 				SCREAMING_SNAKE_CASE     (      __UpperCamelCase :						Dict							,						__UpperCamelCase :						float = 0.0							,						__UpperCamelCase :						float = 1.0       ) ->       int:

				"""simple docstring"""







				return 1 / sqrt(2 * pi * sigma**2       ) * exp(-((x - mu) ** 2) / (2 * sigma**2)       )


if __name__ == "__main__":
					import doctest

					doctest.testmod()