

'''simple docstring'''
import inspect
import tempfile
import unittest

from huggingface_hub import hf_hub_download

from transformers import is_torch_available
from transformers.testing_utils import is_flaky, require_torch, slow, torch_device

from ...test_configuration_common import ConfigTester
from ...test_modeling_common import ModelTesterMixin, floats_tensor, ids_tensor
from ...test_pipeline_mixin import PipelineTesterMixin


lowerCamelCase			:				List[str] 			=					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__   :





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





       '''simple docstring'''






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

       _SCREAMING_SNAKE_CASE							   =context_length
       _SCREAMING_SNAKE_CASE							   =prediction_length + label_length
       _SCREAMING_SNAKE_CASE							   =label_length

       _SCREAMING_SNAKE_CASE							   =moving_average
       _SCREAMING_SNAKE_CASE							   =autocorrelation_factor





  def 						A  (		self				:     List[str]			)     ->   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 						A  (		self				:     int							, _a				:     Union[str, Any]			)     ->   Optional[Any]:





       '''simple docstring'''






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

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

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

       _SCREAMING_SNAKE_CASE							   ={
           'past_values': past_values,
           'static_categorical_features': static_categorical_features,
           'past_time_features': past_time_features,
           'past_observed_mask': past_observed_mask,
           'future_time_features': future_time_features,
           'future_values': future_values,
       }
       return inputs_dict





  def 						A  (		self				:     List[Any]			)     ->   Tuple:





       '''simple docstring'''






       _SCREAMING_SNAKE_CASE							   =self.get_config()
       _SCREAMING_SNAKE_CASE							   =self.prepare_autoformer_inputs_dict(_a			)
       return config, inputs_dict





  def 						A  (		self				:     Tuple			)     ->   Tuple:





       '''simple docstring'''






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





  def 						A  (		self				:     Optional[Any]							, _a				:     List[Any]							, _a				:     Optional[int]			)     ->   Union[str, Any]:





       '''simple docstring'''






       _SCREAMING_SNAKE_CASE							   =AutoformerModel(config=_a			).to(_a			).eval()
       _SCREAMING_SNAKE_CASE							   =model(**_a			)

       _SCREAMING_SNAKE_CASE							   =outputs.encoder_last_hidden_state
       _SCREAMING_SNAKE_CASE							   =outputs.last_hidden_state

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

       _SCREAMING_SNAKE_CASE   ,						_SCREAMING_SNAKE_CASE   ,						_SCREAMING_SNAKE_CASE   ,						_SCREAMING_SNAKE_CASE   ,						_SCREAMING_SNAKE_CASE							   =model.create_network_inputs(**_a			)
       _SCREAMING_SNAKE_CASE   ,						_SCREAMING_SNAKE_CASE							   =model.decomposition_layer(transformer_inputs[:, : config.context_length, ...]			)

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

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

       _SCREAMING_SNAKE_CASE							   =torch.cat(
           (
               torch.cat((seasonal_input[:, -config.label_length :, ...], zeros)							, dim=1			),
               feature[:, config.context_length - config.label_length :, ...],
           )							, dim=-1							, )
       _SCREAMING_SNAKE_CASE							   =torch.cat(
           (
               torch.cat((trend_input[:, -config.label_length :, ...], mean)							, dim=1			),
               feature[:, config.context_length - config.label_length :, ...],
           )							, dim=-1							, )

       with tempfile.TemporaryDirectory() as tmpdirname:
            _SCREAMING_SNAKE_CASE							   =model.get_decoder()
            decoder.save_pretrained(_a			)
            _SCREAMING_SNAKE_CASE							   =AutoformerDecoder.from_pretrained(_a			).to(_a			)

       _SCREAMING_SNAKE_CASE							   =decoder(
           trend=_a							, inputs_embeds=_a							, encoder_hidden_states=_a							, )[0]

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







@require_torch
class 							A__   (	A__					,       A__					,       unittest.TestCase  ):
  A__					=    (AutoformerModel, AutoformerForPrediction) if is_torch_available() else ()
  A__					=    (AutoformerForPrediction,) if is_torch_available() else ()
  A__					=    {'feature-extraction': AutoformerModel} if is_torch_available() else {}
  A__					=    False
  A__					=    False
  A__					=    False
  A__					=    False
  A__					=    False
  A__					=    False





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





       '''simple docstring'''






       _SCREAMING_SNAKE_CASE							   =AutoformerModelTester(self			)
       _SCREAMING_SNAKE_CASE							   =ConfigTester(self							, config_class=_a							, has_text_modality=_a			)





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





       '''simple docstring'''






       self.config_tester.run_common_tests()





  def 						A  (		self				:     Any			)     ->   str:





       '''simple docstring'''






       _SCREAMING_SNAKE_CASE   ,						_SCREAMING_SNAKE_CASE							   =self.model_tester.prepare_config_and_inputs()
       for model_class in self.all_model_classes:
            _SCREAMING_SNAKE_CASE							   =model_class(_a			)

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





  def 						A  (		self				:     Optional[Any]			)     ->   List[str]:





       '''simple docstring'''






       _SCREAMING_SNAKE_CASE							   =self.model_tester.prepare_config_and_inputs_for_common()
       self.model_tester.check_encoder_decoder_model_standalone(*_a			)





  @unittest.skip(reason='Model has no tokens embeddings'			)
  def 						A  (		self				:     List[Any]			)     ->   Optional[int]:





       '''simple docstring'''






       pass





  def 						A  (		self				:     Dict			)     ->   List[Any]:





       '''simple docstring'''






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





  def 						A  (		self				:     str			)     ->   Any:





       '''simple docstring'''






       _SCREAMING_SNAKE_CASE   ,						_SCREAMING_SNAKE_CASE							   =self.model_tester.prepare_config_and_inputs_for_common()

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

            _SCREAMING_SNAKE_CASE							   =[
                'past_values',
                'past_time_features',
                'past_observed_mask',
                'static_categorical_features',
                'static_real_features',
                'future_values',
                'future_time_features',
            ]

            if model.__class__.__name__ in ["AutoformerForPrediction"]:
                 expected_arg_names.append('future_observed_mask'			)

            expected_arg_names.extend(
                [
                    'decoder_attention_mask',
                    'head_mask',
                    'decoder_head_mask',
                    'cross_attn_head_mask',
                    'encoder_outputs',
                    'past_key_values',
                    'output_hidden_states',
                    'output_attentions',
                    'use_cache',
                    'return_dict',
                ]			)

            self.assertListEqual(arg_names[: len(_a			)]							, _a			)





  def 						A  (		self				:     List[Any]			)     ->   Tuple:





       '''simple docstring'''






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

       _SCREAMING_SNAKE_CASE							   =getattr(self.model_tester							, 'seq_length'							, _a			)
       _SCREAMING_SNAKE_CASE							   =getattr(self.model_tester							, 'decoder_seq_length'							, _a			)
       _SCREAMING_SNAKE_CASE							   =getattr(self.model_tester							, 'encoder_seq_length'							, _a			)
       _SCREAMING_SNAKE_CASE							   =getattr(self.model_tester							, 'd_model'							, _a			)
       _SCREAMING_SNAKE_CASE							   =getattr(self.model_tester							, 'num_attention_heads'							, _a			)
       _SCREAMING_SNAKE_CASE							   =d_model // num_attention_heads

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

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

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

            _SCREAMING_SNAKE_CASE							   =7

            if "last_hidden_state" in outputs:
                 correct_outlen += 1

            if "trend" in outputs:
                 correct_outlen += 1

            if "past_key_values" in outputs:
                 correct_outlen += 1  # past_key_values have been returned

            if "loss" in outputs:
                 correct_outlen += 1

            if "params" in outputs:
                 correct_outlen += 1

            self.assertEqual(_a							, _a			)

            # decoder attentions
            _SCREAMING_SNAKE_CASE							   =outputs.decoder_attentions
            self.assertIsInstance(_a							, (list, tuple)			)
            self.assertEqual(len(_a			)							, self.model_tester.num_hidden_layers			)
            self.assertListEqual(
                list(decoder_attentions[0].shape[-3:]			)							, [self.model_tester.num_attention_heads, decoder_seq_length, dim]							, )

            # cross attentions
            _SCREAMING_SNAKE_CASE							   =outputs.cross_attentions
            self.assertIsInstance(_a							, (list, tuple)			)
            self.assertEqual(len(_a			)							, self.model_tester.num_hidden_layers			)
            self.assertListEqual(
                list(cross_attentions[0].shape[-3:]			)							, [self.model_tester.num_attention_heads, decoder_seq_length, dim]							, )

       # Check attention is always last and order is fine
       _SCREAMING_SNAKE_CASE							   =True
       _SCREAMING_SNAKE_CASE							   =True
       _SCREAMING_SNAKE_CASE							   =model_class(_a			)
       model.to(_a			)
       model.eval()
       with torch.no_grad():
            _SCREAMING_SNAKE_CASE							   =model(**self._prepare_for_class(_a							, _a			)			)

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

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

       self.assertEqual(len(_a			)							, self.model_tester.num_hidden_layers			)
       self.assertListEqual(
           list(self_attentions[0].shape[-3:]			)							, [self.model_tester.num_attention_heads, encoder_seq_length, dim]							, )





  @is_flaky()
  def 						A  (		self				:     List[str]			)     ->   Dict:





       '''simple docstring'''






       super().test_retain_grad_hidden_states_attentions()


def 	_lowerCAmelCase (						_UpperCamelCase						:					Optional[int]="train-batch.pt"    )				->      Dict:
     """simple docstring"""




     _SCREAMING_SNAKE_CASE							   =hf_hub_download(repo_id='hf-internal-testing/tourism-monthly-batch'							,			filename=_UpperCamelCase							,			repo_type='dataset'    )
     _SCREAMING_SNAKE_CASE							   =torch.load(_UpperCamelCase							,			map_location=_UpperCamelCase    )
     return batch







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





  def 						A  (		self				:     List[str]			)     ->   int:





       '''simple docstring'''






       _SCREAMING_SNAKE_CASE							   =AutoformerModel.from_pretrained('huggingface/autoformer-tourism-monthly'			).to(_a			)
       _SCREAMING_SNAKE_CASE							   =prepare_batch()

       with torch.no_grad():
            _SCREAMING_SNAKE_CASE							   =model(
                past_values=batch['past_values']							, past_time_features=batch['past_time_features']							, past_observed_mask=batch['past_observed_mask']							, static_categorical_features=batch['static_categorical_features']							, future_values=batch['future_values']							, future_time_features=batch['future_time_features']							, )[0]

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

       _SCREAMING_SNAKE_CASE							   =torch.tensor(
           [[0.35_93, -1.33_98, 0.63_30], [0.22_79, 1.53_96, -0.17_92], [0.04_50, 1.32_25, -0.23_35]]							, device=_a			)
       self.assertTrue(torch.allclose(output[0, :3, :3]							, _a							, atol=_a			)			)





  def 						A  (		self				:     Optional[Any]			)     ->   str:





       '''simple docstring'''






       _SCREAMING_SNAKE_CASE							   =AutoformerForPrediction.from_pretrained('huggingface/autoformer-tourism-monthly'			).to(_a			)
       _SCREAMING_SNAKE_CASE							   =prepare_batch('val-batch.pt'			)
       with torch.no_grad():
            _SCREAMING_SNAKE_CASE							   =model(
                past_values=batch['past_values']							, past_time_features=batch['past_time_features']							, past_observed_mask=batch['past_observed_mask']							, static_categorical_features=batch['static_categorical_features']							, ).encoder_last_hidden_state
       _SCREAMING_SNAKE_CASE							   =torch.Size((64, model.config.context_length, model.config.d_model)			)
       self.assertEqual(output.shape							, _a			)

       _SCREAMING_SNAKE_CASE							   =torch.tensor(
           [[-0.07_34, -0.90_36, 0.83_58], [4.71_86, 2.41_13, 1.95_81], [1.79_53, 2.35_58, 1.29_70]]							, device=_a			)
       self.assertTrue(torch.allclose(output[0, :3, :3]							, _a							, atol=_a			)			)





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





       '''simple docstring'''






       _SCREAMING_SNAKE_CASE							   =AutoformerForPrediction.from_pretrained('huggingface/autoformer-tourism-monthly'			).to(_a			)
       _SCREAMING_SNAKE_CASE							   =prepare_batch('val-batch.pt'			)
       with torch.no_grad():
            _SCREAMING_SNAKE_CASE							   =model.generate(
                static_categorical_features=batch['static_categorical_features']							, past_time_features=batch['past_time_features']							, past_values=batch['past_values']							, future_time_features=batch['future_time_features']							, past_observed_mask=batch['past_observed_mask']							, )
       _SCREAMING_SNAKE_CASE							   =torch.Size((64, model.config.num_parallel_samples, model.config.prediction_length)			)
       self.assertEqual(outputs.sequences.shape							, _a			)

       _SCREAMING_SNAKE_CASE							   =torch.tensor([31_30.67_63, 40_56.52_93, 70_53.07_86]							, device=_a			)
       _SCREAMING_SNAKE_CASE							   =outputs.sequences.mean(dim=1			)
       self.assertTrue(torch.allclose(mean_prediction[0, -3:]							, _a							, rtol=1e-1			)			)



'''simple docstring'''
def 	_lowerCAmelCase (						_UpperCamelCase						:					float							,			_UpperCamelCase						:					float    )				->      float:
     """simple docstring"""




     if mass < 0:
          raise ValueError('The mass of a body cannot be negative'    )
     return 0.5 * mass * abs(_UpperCamelCase    ) * abs(_UpperCamelCase    )


if __name__ == "__main__":
     import doctest

     doctest.testmod(verbose=True)
