





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


A__:  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     _a  :

      """simple docstring"""


      def __init__(						self:       Optional[Any]     ,  __lowerCamelCase:       List[Any]     ,  __lowerCamelCase:       Tuple=16     ,  __lowerCamelCase:       Optional[Any]=13     ,  __lowerCamelCase:       List[Any]=7     ,  __lowerCamelCase:       Dict=14     ,  __lowerCamelCase:       Dict=10     ,  __lowerCamelCase:       Optional[int]=19     ,  __lowerCamelCase:       Dict=5     ,  __lowerCamelCase:       Tuple=4     ,  __lowerCamelCase:       Optional[int]=True     ,  __lowerCamelCase:       List[str]=16     ,  __lowerCamelCase:       List[str]=2     ,  __lowerCamelCase:       Optional[Any]=4     ,  __lowerCamelCase:       Optional[int]=4     ,  __lowerCamelCase:       Tuple="gelu"     ,  __lowerCamelCase:       Tuple=0.1     ,  __lowerCamelCase:       Optional[Any]=0.1     ,  __lowerCamelCase:       Dict=[1, 2, 3, 4, 5]     ,  __lowerCamelCase:       int=25     ,  __lowerCamelCase:       Dict=5     ,  ):
          '''simple docstring'''


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

          UpperCamelCase__:     Optional[Any]     							=			context_length
          UpperCamelCase__:     Optional[Any]     							=			prediction_length + label_length
          UpperCamelCase__:     List[Any]     							=			label_length

          UpperCamelCase__:     Tuple     							=			moving_average
          UpperCamelCase__:     Union[str, Any]     							=			autocorrelation_factor






      def        UpperCAmelCase_    (						self:       List[Any] ):
          '''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        UpperCAmelCase_    (						self:       List[str]     ,  __lowerCamelCase:       Tuple ):
          '''simple docstring'''


          UpperCamelCase__:     Optional[int]     							=			config.context_length + max(config.lags_sequence )

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

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

          UpperCamelCase__:     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        UpperCAmelCase_    (						self:       List[str] ):
          '''simple docstring'''


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






      def        UpperCAmelCase_    (						self:       List[str] ):
          '''simple docstring'''


          UpperCamelCase__:     List[Any]     							=			self.prepare_config_and_inputs()
          return config, inputs_dict






      def        UpperCAmelCase_    (						self:       int     ,  __lowerCamelCase:       Union[str, Any]     ,  __lowerCamelCase:       Any ):
          '''simple docstring'''


          UpperCamelCase__:     Dict     							=			AutoformerModel(config=SCREAMING_SNAKE_CASE__ ).to(SCREAMING_SNAKE_CASE__ ).eval()
          UpperCamelCase__:     Optional[int]     							=			model(**SCREAMING_SNAKE_CASE__ )

          UpperCamelCase__:     Optional[int]     							=			outputs.encoder_last_hidden_state
          UpperCamelCase__:     Optional[int]     							=			outputs.last_hidden_state

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

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

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

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

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

          UpperCamelCase__:     str     							=			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     _a  (  a__		,       a__		,       unittest.TestCase):

      """simple docstring"""


      UpperCamelCase__   						=    (AutoformerModel, AutoformerForPrediction) if is_torch_available() else ()
      UpperCamelCase__   						=    (AutoformerForPrediction,) if is_torch_available() else ()
      UpperCamelCase__   						=    {'''feature-extraction''': AutoformerModel} if is_torch_available() else {}
      UpperCamelCase__   						=    False
      UpperCamelCase__   						=    False
      UpperCamelCase__   						=    False
      UpperCamelCase__   						=    False
      UpperCamelCase__   						=    False
      UpperCamelCase__   						=    False






      def        UpperCAmelCase_    (						self:       Dict ):
          '''simple docstring'''


          UpperCamelCase__:     Optional[int]     							=			AutoformerModelTester(self )
          UpperCamelCase__:     List[str]     							=			ConfigTester(self     ,  config_class=SCREAMING_SNAKE_CASE__     ,  has_text_modality=SCREAMING_SNAKE_CASE__ )






      def        UpperCAmelCase_    (						self:       int ):
          '''simple docstring'''


          self.config_tester.run_common_tests()






      def        UpperCAmelCase_    (						self:       Dict ):
          '''simple docstring'''


          UpperCamelCase__:     Optional[Any]     							=			self.model_tester.prepare_config_and_inputs()
          for model_class in self.all_model_classes:
              UpperCamelCase__:     Optional[int]     							=			model_class(SCREAMING_SNAKE_CASE__ )

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






      def        UpperCAmelCase_    (						self:       Tuple ):
          '''simple docstring'''


          UpperCamelCase__:     str     							=			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        UpperCAmelCase_    (						self:       Tuple ):
          '''simple docstring'''


          pass






      def        UpperCAmelCase_    (						self:       Optional[Any] ):
          '''simple docstring'''


          UpperCamelCase__:     str     							=			inspect.signature(getattr(SCREAMING_SNAKE_CASE__     ,  "forward" ) )
          # The main input is the name of the argument after `self`
          UpperCamelCase__:     Dict     							=			list(model_signature.parameters.keys() )[1]
          self.assertEqual(AutoformerModel.main_input_name     ,  SCREAMING_SNAKE_CASE__ )






      def        UpperCAmelCase_    (						self:       Any ):
          '''simple docstring'''


          UpperCamelCase__:     Tuple     							=			self.model_tester.prepare_config_and_inputs_for_common()

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

              UpperCamelCase__:     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(SCREAMING_SNAKE_CASE__ )]     ,  SCREAMING_SNAKE_CASE__ )






      def        UpperCAmelCase_    (						self:       Tuple ):
          '''simple docstring'''


          UpperCamelCase__:     List[Any]     							=			self.model_tester.prepare_config_and_inputs_for_common()
          UpperCamelCase__:     Optional[int]     							=			True

          UpperCamelCase__:     List[Any]     							=			getattr(self.model_tester     ,  "seq_length"     ,  SCREAMING_SNAKE_CASE__ )
          UpperCamelCase__:     Dict     							=			getattr(self.model_tester     ,  "decoder_seq_length"     ,  SCREAMING_SNAKE_CASE__ )
          UpperCamelCase__:     Optional[Any]     							=			getattr(self.model_tester     ,  "encoder_seq_length"     ,  SCREAMING_SNAKE_CASE__ )
          UpperCamelCase__:     Any     							=			getattr(self.model_tester     ,  "d_model"     ,  SCREAMING_SNAKE_CASE__ )
          UpperCamelCase__:     Union[str, Any]     							=			getattr(self.model_tester     ,  "num_attention_heads"     ,  SCREAMING_SNAKE_CASE__ )
          UpperCamelCase__:     str     							=			d_model // num_attention_heads

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

              UpperCamelCase__:     str     							=			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
              UpperCamelCase__:     Union[str, Any]     							=			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
              UpperCamelCase__:     List[str]     							=			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
          UpperCamelCase__:     Dict     							=			True
          UpperCamelCase__:     Union[str, Any]     							=			True
          UpperCamelCase__:     str     							=			model_class(SCREAMING_SNAKE_CASE__ )
          model.to(SCREAMING_SNAKE_CASE__ )
          model.eval()
          with torch.no_grad():
              UpperCamelCase__:     str     							=			model(**self._prepare_for_class(SCREAMING_SNAKE_CASE__     ,  SCREAMING_SNAKE_CASE__ ) )

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

          UpperCamelCase__:     Any     							=			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        UpperCAmelCase_    (						self:       Dict ):
          '''simple docstring'''


          super().test_retain_grad_hidden_states_attentions()



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




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

      """simple docstring"""


      def        UpperCAmelCase_    (						self:       int ):
          '''simple docstring'''


          UpperCamelCase__:     Union[str, Any]     							=			AutoformerModel.from_pretrained("huggingface/autoformer-tourism-monthly" ).to(SCREAMING_SNAKE_CASE__ )
          UpperCamelCase__:     Any     							=			prepare_batch()

          with torch.no_grad():
              UpperCamelCase__:     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"]     ,  future_values=batch["future_values"]     ,  future_time_features=batch["future_time_features"]     ,  )[0]

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

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






      def        UpperCAmelCase_    (						self:       Optional[int] ):
          '''simple docstring'''


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

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






      def        UpperCAmelCase_    (						self:       Optional[Any] ):
          '''simple docstring'''


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

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



"""simple docstring"""




def lowercase_  (   _snake_case						):
							if a < 0:
														raise ValueError("""Input value must be a positive integer"""						)
							elif isinstance(_snake_case   ,_snake_case						):
														raise TypeError("""Input value must be a 'int' type"""						)
							return bin(_snake_case						).count("""1"""						)


if __name__ == "__main__":
					import doctest

					doctest.testmod()




