




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


lowercase      :   Optional[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 					_a :





     '''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							,)      ->				Dict:
          snake_case			:      Any        =			d_model
          snake_case			:      Optional[int]        =			parent
          snake_case			:      List[Any]        =			batch_size
          snake_case			:      List[str]        =			prediction_length
          snake_case			:      List[str]        =			context_length
          snake_case			:      int        =			cardinality
          snake_case			:      int        =			num_time_features
          snake_case			:      int        =			lags_sequence
          snake_case			:      Optional[int]        =			embedding_dimension
          snake_case			:      Dict        =			is_training
          snake_case			:      Any        =			hidden_size
          snake_case			:      int        =			num_hidden_layers
          snake_case			:      int        =			num_attention_heads
          snake_case			:      List[str]        =			intermediate_size
          snake_case			:      Optional[int]        =			hidden_act
          snake_case			:      Dict        =			hidden_dropout_prob
          snake_case			:      List[Any]        =			attention_probs_dropout_prob

          snake_case			:      str        =			context_length
          snake_case			:      Optional[Any]        =			prediction_length + label_length
          snake_case			:      Optional[Any]        =			label_length

          snake_case			:      Dict        =			moving_average
          snake_case			:      List[Any]        =			autocorrelation_factor







     def 			snake_case_      (				self							)      ->				Optional[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 			snake_case_      (				self							,__a							)      ->				Optional[Any]:
          snake_case			:      str        =			config.context_length + max(config.lags_sequence							)

          snake_case			:      List[str]        =			ids_tensor([self.batch_size, 1]							,config.cardinality[0]							)
          snake_case			:      Any        =			floats_tensor([self.batch_size, _past_length, config.num_time_features]							)
          snake_case			:      Dict        =			floats_tensor([self.batch_size, _past_length]							)
          snake_case			:      str        =			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			:      Tuple        =			floats_tensor([self.batch_size, config.prediction_length]							)

          snake_case			:      Optional[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 			snake_case_      (				self							)      ->				int:
          snake_case			:      List[str]        =			self.get_config()
          snake_case			:      str        =			self.prepare_autoformer_inputs_dict(__a							)
          return config, inputs_dict







     def 			snake_case_      (				self							)      ->				str:
          snake_case ,		snake_case			:      List[str]        =			self.prepare_config_and_inputs()
          return config, inputs_dict







     def 			snake_case_      (				self							,__a							,__a							)      ->				str:
          snake_case			:      Optional[int]        =			AutoformerModel(config=__a							).to(__a							).eval()
          snake_case			:      List[str]        =			model(**__a							)

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

          with tempfile.TemporaryDirectory() as tmpdirname:
               snake_case			:      List[Any]        =			model.get_encoder()
               encoder.save_pretrained(__a							)
               snake_case			:      Dict        =			AutoformerEncoder.from_pretrained(__a							).to(__a							)

          snake_case ,		snake_case ,		snake_case ,		snake_case ,		snake_case			:      Tuple        =			model.create_network_inputs(**__a							)
          snake_case ,		snake_case			:      int        =			model.decomposition_layer(transformer_inputs[:, : config.context_length, ...]							)

          snake_case			:      Union[str, Any]        =			torch.cat(
              (transformer_inputs[:, : config.context_length, ...], feature[:, : config.context_length, ...])							,dim=-1							,)
          snake_case			:      Optional[int]        =			encoder(inputs_embeds=__a							)[0]
          self.parent.assertTrue((encoder_last_hidden_state_a - encoder_last_hidden_state).abs().max().item() < 1E-3							)

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

          snake_case			:      Optional[int]        =			torch.cat(
              (
                  torch.cat((seasonal_input[:, -config.label_length :, ...], zeros)							,dim=1							),
                  feature[:, config.context_length - config.label_length :, ...],
              )							,dim=-1							,)
          snake_case			:      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(__a							)
               snake_case			:      Tuple        =			AutoformerDecoder.from_pretrained(__a							).to(__a							)

          snake_case			:      Optional[int]        =			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 					_a (a__,		a__,		unittest.TestCase	):





     '''simple docstring'''



     lowerCAmelCase_  :     Tuple          =					(AutoformerModel, AutoformerForPrediction) if is_torch_available() else ()
     lowerCAmelCase_  :     str          =					(AutoformerForPrediction,) if is_torch_available() else ()
     lowerCAmelCase_  :     Optional[Any]          =					{"""feature-extraction""": AutoformerModel} if is_torch_available() else {}
     lowerCAmelCase_  :     int          =					False
     lowerCAmelCase_  :     int          =					False
     lowerCAmelCase_  :     Optional[Any]          =					False
     lowerCAmelCase_  :     Dict          =					False
     lowerCAmelCase_  :     Any          =					False
     lowerCAmelCase_  :     Optional[Any]          =					False







     def 			snake_case_      (				self							)      ->				Dict:
          snake_case			:      int        =			AutoformerModelTester(self							)
          snake_case			:      List[str]        =			ConfigTester(self							,config_class=__a							,has_text_modality=__a							)







     def 			snake_case_      (				self							)      ->				Any:
          self.config_tester.run_common_tests()







     def 			snake_case_      (				self							)      ->				str:
          snake_case ,		snake_case			:      Optional[Any]        =			self.model_tester.prepare_config_and_inputs()
          for model_class in self.all_model_classes:
               snake_case			:      Union[str, Any]        =			model_class(__a							)

               with tempfile.TemporaryDirectory() as tmpdirname:
                    model.save_pretrained(__a							)
                    snake_case ,		snake_case			:      Optional[int]        =			model_class.from_pretrained(__a							,output_loading_info=__a							)
               self.assertEqual(info["""missing_keys"""]							,[]							)







     def 			snake_case_      (				self							)      ->				Tuple:
          snake_case			:      Tuple        =			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 			snake_case_      (				self							)      ->				Union[str, Any]:
          pass







     def 			snake_case_      (				self							)      ->				str:
          snake_case			:      List[Any]        =			inspect.signature(getattr(__a							,"""forward"""							)							)
          # The main input is the name of the argument after `self`
          snake_case			:      List[str]        =			list(model_signature.parameters.keys()							)[1]
          self.assertEqual(AutoformerModel.main_input_name							,__a							)







     def 			snake_case_      (				self							)      ->				Tuple:
          snake_case ,		snake_case			:      Union[str, Any]        =			self.model_tester.prepare_config_and_inputs_for_common()

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

               snake_case			:      Optional[Any]        =			[
                   """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 			snake_case_      (				self							)      ->				List[str]:
          snake_case ,		snake_case			:      Any        =			self.model_tester.prepare_config_and_inputs_for_common()
          snake_case			:      Any        =			True

          snake_case			:      int        =			getattr(self.model_tester							,"""seq_length"""							,__a							)
          snake_case			:      Any        =			getattr(self.model_tester							,"""decoder_seq_length"""							,__a							)
          snake_case			:      str        =			getattr(self.model_tester							,"""encoder_seq_length"""							,__a							)
          snake_case			:      List[Any]        =			getattr(self.model_tester							,"""d_model"""							,__a							)
          snake_case			:      Optional[int]        =			getattr(self.model_tester							,"""num_attention_heads"""							,__a							)
          snake_case			:      Dict        =			d_model // num_attention_heads

          for model_class in self.all_model_classes:
               snake_case			:      List[str]        =			True
               snake_case			:      List[Any]        =			False
               snake_case			:      Any        =			True
               snake_case			:      List[str]        =			model_class(__a							)
               model.to(__a							)
               model.eval()
               with torch.no_grad():
                    snake_case			:      List[Any]        =			model(**self._prepare_for_class(__a							,__a							)							)
               snake_case			:      Union[str, Any]        =			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"]
               snake_case			:      List[Any]        =			True
               snake_case			:      Any        =			model_class(__a							)
               model.to(__a							)
               model.eval()
               with torch.no_grad():
                    snake_case			:      Optional[Any]        =			model(**self._prepare_for_class(__a							,__a							)							)
               snake_case			:      Union[str, Any]        =			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]							,)
               snake_case			:      List[Any]        =			len(__a							)

               snake_case			:      Tuple        =			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
               snake_case			:      Optional[Any]        =			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
               snake_case			:      Optional[Any]        =			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
          snake_case			:      Tuple        =			True
          snake_case			:      List[str]        =			True
          snake_case			:      int        =			model_class(__a							)
          model.to(__a							)
          model.eval()
          with torch.no_grad():
               snake_case			:      Dict        =			model(**self._prepare_for_class(__a							,__a							)							)

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

          snake_case			:      Optional[Any]        =			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 			snake_case_      (				self							)      ->				List[str]:
          super().test_retain_grad_hidden_states_attentions()



def        lowerCamelCase__				(      __lowercase="train-batch.pt"				):
     snake_case			:      Optional[Any]        =			hf_hub_download(repo_id="""hf-internal-testing/tourism-monthly-batch"""     ,   filename=__lowercase     ,   repo_type="""dataset"""				)
     snake_case			:      List[Any]        =			torch.load(__lowercase     ,   map_location=__lowercase				)
     return batch







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





     '''simple docstring'''



     def 			snake_case_      (				self							)      ->				Dict:
          snake_case			:      int        =			AutoformerModel.from_pretrained("""huggingface/autoformer-tourism-monthly"""							).to(__a							)
          snake_case			:      Optional[int]        =			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			:      Dict        =			torch.Size(
              (64, model.config.prediction_length + model.config.label_length, model.config.feature_size)							)
          self.assertEqual(output.shape							,__a							)

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







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

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







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

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









'''simple docstring'''







def        lowerCamelCase__				(      __lowercase				):
     if not isinstance(__lowercase     ,   __lowercase				):
          snake_case			:      int        =			F'''Input value of [number={number}] must be an integer'''
          raise TypeError(__lowercase				)
     if number < 0:
          return False
     snake_case			:      Optional[int]        =			number * number
     while number > 0:
          if number % 10 != number_square % 10:
               return False
          number //= 10
          number_square //= 10
     return True


if __name__ == "__main__":
      import doctest

      doctest.testmod()



