





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_    =				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					,							__magic_name__					,							__magic_name__=16					,							__magic_name__=13					,							__magic_name__=7					,							__magic_name__=14					,							__magic_name__=10					,							__magic_name__=19					,							__magic_name__=5					,							__magic_name__=4					,							__magic_name__=True					,							__magic_name__=16					,							__magic_name__=2					,							__magic_name__=4					,							__magic_name__=4					,							__magic_name__="gelu"					,							__magic_name__=0.1					,							__magic_name__=0.1					,							__magic_name__=[1, 2, 3, 4, 5]					,							__magic_name__=25					,							__magic_name__=5					,							)     ->   Optional[int]:


							'''simple docstring'''



							snake_case_						:    Dict				=  d_model
							snake_case_						:    Optional[Any]				=  parent
							snake_case_						:    List[Any]				=  batch_size
							snake_case_						:    Dict				=  prediction_length
							snake_case_						:    Optional[Any]				=  context_length
							snake_case_						:    Any				=  cardinality
							snake_case_						:    int				=  num_time_features
							snake_case_						:    Optional[int]				=  lags_sequence
							snake_case_						:    Optional[Any]				=  embedding_dimension
							snake_case_						:    Any				=  is_training
							snake_case_						:    List[Any]				=  hidden_size
							snake_case_						:    Dict				=  num_hidden_layers
							snake_case_						:    str				=  num_attention_heads
							snake_case_						:    int				=  intermediate_size
							snake_case_						:    List[str]				=  hidden_act
							snake_case_						:    int				=  hidden_dropout_prob
							snake_case_						:    List[str]				=  attention_probs_dropout_prob

							snake_case_						:    str				=  context_length
							snake_case_						:    Tuple				=  prediction_length + label_length
							snake_case_						:    int				=  label_length

							snake_case_						:    Optional[int]				=  moving_average
							snake_case_						:    Optional[int]				=  autocorrelation_factor






			def 			lowerCamelCase  (self    )     ->   Dict:


							'''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 			lowerCamelCase  (self					,							__magic_name__    )     ->   Dict:


							'''simple docstring'''



							snake_case_						:    Tuple				=  config.context_length + max(config.lags_sequence    )

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

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

							snake_case_						:    int				=  {
							    '''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 			lowerCamelCase  (self    )     ->   Optional[Any]:


							'''simple docstring'''



							snake_case_						:    Optional[int]				=  self.get_config()
							snake_case_						:    Dict				=  self.prepare_autoformer_inputs_dict(__magic_name__    )
							return config, inputs_dict






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


							'''simple docstring'''



							snake_case_			,							snake_case_						:    Optional[Any]				=  self.prepare_config_and_inputs()
							return config, inputs_dict






			def 			lowerCamelCase  (self					,							__magic_name__					,							__magic_name__    )     ->   int:


							'''simple docstring'''



							snake_case_						:    List[str]				=  AutoformerModel(config=__magic_name__    ).to(__magic_name__    ).eval()
							snake_case_						:    Union[str, Any]				=  model(**__magic_name__    )

							snake_case_						:    Optional[Any]				=  outputs.encoder_last_hidden_state
							snake_case_						:    Any				=  outputs.last_hidden_state

							with tempfile.TemporaryDirectory() as tmpdirname:
											snake_case_						:    Dict				=  model.get_encoder()
											encoder.save_pretrained(__magic_name__    )
											snake_case_						:    Optional[Any]				=  AutoformerEncoder.from_pretrained(__magic_name__    ).to(__magic_name__    )

							snake_case_			,							snake_case_			,							snake_case_			,							snake_case_			,							snake_case_						:    Union[str, Any]				=  model.create_network_inputs(**__magic_name__    )
							snake_case_			,							snake_case_						:    Any				=  model.decomposition_layer(transformer_inputs[:, : config.context_length, ...]    )

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

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

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

							snake_case_						:    Any				=  decoder(
							    trend=__magic_name__					,							inputs_embeds=__magic_name__					,							encoder_hidden_states=__magic_name__					,							)[0]

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




@require_torch
class 					__lowerCAmelCase       (			_a,     _a,     unittest.TestCase     ):
			lowerCamelCase_     :     List[Any]       			=      (AutoformerModel, AutoformerForPrediction) if is_torch_available() else ()
			lowerCamelCase_     :     str       			=      (AutoformerForPrediction,) if is_torch_available() else ()
			lowerCamelCase_     :     Dict       			=      {'''feature-extraction''': AutoformerModel} if is_torch_available() else {}
			lowerCamelCase_     :     Union[str, Any]       			=      False
			lowerCamelCase_     :     str       			=      False
			lowerCamelCase_     :     List[Any]       			=      False
			lowerCamelCase_     :     Union[str, Any]       			=      False
			lowerCamelCase_     :     Optional[int]       			=      False
			lowerCamelCase_     :     Tuple       			=      False






			def 			lowerCamelCase  (self    )     ->   int:


							'''simple docstring'''



							snake_case_						:    str				=  AutoformerModelTester(self    )
							snake_case_						:    List[str]				=  ConfigTester(self					,							config_class=__magic_name__					,							has_text_modality=__magic_name__    )






			def 			lowerCamelCase  (self    )     ->   List[Any]:


							'''simple docstring'''



							self.config_tester.run_common_tests()






			def 			lowerCamelCase  (self    )     ->   List[Any]:


							'''simple docstring'''



							snake_case_			,							snake_case_						:    Optional[int]				=  self.model_tester.prepare_config_and_inputs()
							for model_class in self.all_model_classes:
											snake_case_						:    Any				=  model_class(__magic_name__    )

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






			def 			lowerCamelCase  (self    )     ->   Optional[int]:


							'''simple docstring'''



							snake_case_						:    Optional[Any]				=  self.model_tester.prepare_config_and_inputs_for_common()
							self.model_tester.check_encoder_decoder_model_standalone(*__magic_name__    )






			@unittest.skip(reason='''Model has no tokens embeddings'''    )
			def 			lowerCamelCase  (self    )     ->   Any:


							'''simple docstring'''



							pass






			def 			lowerCamelCase  (self    )     ->   List[Any]:


							'''simple docstring'''



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






			def 			lowerCamelCase  (self    )     ->   Optional[Any]:


							'''simple docstring'''



							snake_case_			,							snake_case_						:    List[str]				=  self.model_tester.prepare_config_and_inputs_for_common()

							for model_class in self.all_model_classes:
											snake_case_						:    Optional[int]				=  model_class(__magic_name__    )
											snake_case_						:    Any				=  inspect.signature(model.forward    )
											# signature.parameters is an OrderedDict => so arg_names order is deterministic
											snake_case_						:    Union[str, Any]				=  [*signature.parameters.keys()]

											snake_case_						:    Tuple				=  [
											    '''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(__magic_name__    )]					,							__magic_name__    )






			def 			lowerCamelCase  (self    )     ->   str:


							'''simple docstring'''



							snake_case_			,							snake_case_						:    Any				=  self.model_tester.prepare_config_and_inputs_for_common()
							snake_case_						:    Union[str, Any]				=  True

							snake_case_						:    Union[str, Any]				=  getattr(self.model_tester					,							'''seq_length'''					,							__magic_name__    )
							snake_case_						:    Optional[int]				=  getattr(self.model_tester					,							'''decoder_seq_length'''					,							__magic_name__    )
							snake_case_						:    Optional[Any]				=  getattr(self.model_tester					,							'''encoder_seq_length'''					,							__magic_name__    )
							snake_case_						:    Tuple				=  getattr(self.model_tester					,							'''d_model'''					,							__magic_name__    )
							snake_case_						:    List[str]				=  getattr(self.model_tester					,							'''num_attention_heads'''					,							__magic_name__    )
							snake_case_						:    int				=  d_model // num_attention_heads

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

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

											self.assertListEqual(
											    list(attentions[0].shape[-3:]    )					,							[self.model_tester.num_attention_heads, encoder_seq_length, dim]					,							)
											snake_case_						:    List[str]				=  len(__magic_name__    )

											snake_case_						:    Union[str, 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(__magic_name__					,							__magic_name__    )

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

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

							snake_case_						:    Optional[Any]				=  outputs.encoder_attentions if config.is_encoder_decoder else outputs.attentions

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


							'''simple docstring'''



							super().test_retain_grad_hidden_states_attentions()






def      lowerCamelCase_						(				_UpperCamelCase="train-batch.pt" )		->						Optional[Any]:






				"""simple docstring"""


				snake_case_						:    str				=  hf_hub_download(repo_id='''hf-internal-testing/tourism-monthly-batch''' ,				filename=_UpperCamelCase ,				repo_type='''dataset''' )
				snake_case_						:    List[str]				=  torch.load(_UpperCamelCase ,				map_location=_UpperCamelCase )
				return batch





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






			def 			lowerCamelCase  (self    )     ->   int:


							'''simple docstring'''



							snake_case_						:    Dict				=  AutoformerModel.from_pretrained('''huggingface/autoformer-tourism-monthly'''    ).to(__magic_name__    )
							snake_case_						:    Union[str, Any]				=  prepare_batch()

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

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

							snake_case_						:    Optional[int]				=  torch.tensor(
							    [[0.3_593, -1.3_398, 0.6_330], [0.2_279, 1.5_396, -0.1_792], [0.0_450, 1.3_225, -0.2_335]]					,							device=__magic_name__    )
							self.assertTrue(torch.allclose(output[0, :3, :3]					,							__magic_name__					,							atol=__magic_name__    )    )






			def 			lowerCamelCase  (self    )     ->   Optional[int]:


							'''simple docstring'''



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

							snake_case_						:    str				=  torch.tensor(
							    [[-0.0_734, -0.9_036, 0.8_358], [4.7_186, 2.4_113, 1.9_581], [1.7_953, 2.3_558, 1.2_970]]					,							device=__magic_name__    )
							self.assertTrue(torch.allclose(output[0, :3, :3]					,							__magic_name__					,							atol=__magic_name__    )    )






			def 			lowerCamelCase  (self    )     ->   str:


							'''simple docstring'''



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

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







from ...utils import (
    OptionalDependencyNotAvailable,
    is_torch_available,
    is_transformers_available,
    is_transformers_version,
)


try:
	if not (is_transformers_available() and is_torch_available() and is_transformers_version('''>=''', '''4.25.0''')):
		raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
	from ...utils.dummy_torch_and_transformers_objects import UnCLIPImageVariationPipeline, UnCLIPPipeline
else:
	from .pipeline_unclip import UnCLIPPipeline
	from .pipeline_unclip_image_variation import UnCLIPImageVariationPipeline
	from .text_proj import UnCLIPTextProjModel
