

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			:					Optional[Any]						=			1e-4

if is_torch_available():
    import torch

    from transformers import AutoformerConfig, AutoformerForPrediction, AutoformerModel
    from transformers.models.autoformer.modeling_autoformer import AutoformerDecoder, AutoformerEncoder





@require_torch
class 			_SCREAMING_SNAKE_CASE :







   '''simple docstring'''




   def __init__(       self					:			Union[str, Any]     , __lowerCamelCase					:			List[str]     , __lowerCamelCase					:			List[str]=16     , __lowerCamelCase					:			int=13     , __lowerCamelCase					:			Any=7     , __lowerCamelCase					:			List[str]=14     , __lowerCamelCase					:			str=10     , __lowerCamelCase					:			Dict=19     , __lowerCamelCase					:			List[str]=5     , __lowerCamelCase					:			Any=4     , __lowerCamelCase					:			Any=True     , __lowerCamelCase					:			Union[str, Any]=16     , __lowerCamelCase					:			Optional[Any]=2     , __lowerCamelCase					:			Optional[Any]=4     , __lowerCamelCase					:			List[str]=4     , __lowerCamelCase					:			Optional[Any]="gelu"     , __lowerCamelCase					:			Optional[int]=0.1     , __lowerCamelCase					:			Tuple=0.1     , __lowerCamelCase					:			str=[1, 2, 3, 4, 5]     , __lowerCamelCase					:			List[Any]=25     , __lowerCamelCase					:			Any=5     , ):
       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      _snake_case		(       self					:			Dict			):
       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					:			Dict     , __lowerCamelCase					:			Dict			):
       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      _snake_case		(       self					:			Optional[Any]			):
       SCREAMING_SNAKE_CASE										=   self.get_config()
       SCREAMING_SNAKE_CASE										=   self.prepare_autoformer_inputs_dict(__lowerCamelCase			)
       return config, inputs_dict





   def      _snake_case		(       self					:			int			):
       SCREAMING_SNAKE_CASE							,	SCREAMING_SNAKE_CASE										=   self.prepare_config_and_inputs()
       return config, inputs_dict





   def      _snake_case		(       self					:			int     , __lowerCamelCase					:			int     , __lowerCamelCase					:			Dict			):
       SCREAMING_SNAKE_CASE										=   AutoformerModel(config=__lowerCamelCase			).to(__lowerCamelCase			).eval()
       SCREAMING_SNAKE_CASE										=   model(**__lowerCamelCase			)

       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(__lowerCamelCase			)
           SCREAMING_SNAKE_CASE										=   AutoformerEncoder.from_pretrained(__lowerCamelCase			).to(__lowerCamelCase			)

       SCREAMING_SNAKE_CASE							,	SCREAMING_SNAKE_CASE							,	SCREAMING_SNAKE_CASE							,	SCREAMING_SNAKE_CASE							,	SCREAMING_SNAKE_CASE										=   model.create_network_inputs(**__lowerCamelCase			)
       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=__lowerCamelCase			)[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(__lowerCamelCase			)
           SCREAMING_SNAKE_CASE										=   AutoformerDecoder.from_pretrained(__lowerCamelCase			).to(__lowerCamelCase			)

       SCREAMING_SNAKE_CASE										=   decoder(
           trend=__lowerCamelCase     , inputs_embeds=__lowerCamelCase     , encoder_hidden_states=__lowerCamelCase     , )[0]

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





@require_torch
class 			_SCREAMING_SNAKE_CASE (   __snake_case					,       __snake_case					,       unittest.TestCase		):







   '''simple docstring'''




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





   def      _snake_case		(       self					:			Optional[int]			):
       SCREAMING_SNAKE_CASE										=   AutoformerModelTester(self			)
       SCREAMING_SNAKE_CASE										=   ConfigTester(self     , config_class=__lowerCamelCase     , has_text_modality=__lowerCamelCase			)





   def      _snake_case		(       self					:			int			):
       self.config_tester.run_common_tests()





   def      _snake_case		(       self					:			Union[str, Any]			):
       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(__lowerCamelCase			)

           with tempfile.TemporaryDirectory() as tmpdirname:
               model.save_pretrained(__lowerCamelCase			)
               SCREAMING_SNAKE_CASE							,	SCREAMING_SNAKE_CASE										=   model_class.from_pretrained(__lowerCamelCase     , output_loading_info=__lowerCamelCase			)
           self.assertEqual(info["missing_keys"]     , []			)





   def      _snake_case		(       self					:			str			):
       SCREAMING_SNAKE_CASE										=   self.model_tester.prepare_config_and_inputs_for_common()
       self.model_tester.check_encoder_decoder_model_standalone(*__lowerCamelCase			)





   @unittest.skip(reason="Model has no tokens embeddings"			)
   def      _snake_case		(       self					:			Tuple			):
       pass





   def      _snake_case		(       self					:			Dict			):
       SCREAMING_SNAKE_CASE										=   inspect.signature(getattr(__lowerCamelCase     , "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     , __lowerCamelCase			)





   def      _snake_case		(       self					:			Tuple			):
       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(__lowerCamelCase			)
           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(__lowerCamelCase			)]     , __lowerCamelCase			)





   def      _snake_case		(       self					:			str			):
       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"     , __lowerCamelCase			)
       SCREAMING_SNAKE_CASE										=   getattr(self.model_tester     , "decoder_seq_length"     , __lowerCamelCase			)
       SCREAMING_SNAKE_CASE										=   getattr(self.model_tester     , "encoder_seq_length"     , __lowerCamelCase			)
       SCREAMING_SNAKE_CASE										=   getattr(self.model_tester     , "d_model"     , __lowerCamelCase			)
       SCREAMING_SNAKE_CASE										=   getattr(self.model_tester     , "num_attention_heads"     , __lowerCamelCase			)
       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(__lowerCamelCase			)
           model.to(__lowerCamelCase			)
           model.eval()
           with torch.no_grad():
               SCREAMING_SNAKE_CASE										=   model(**self._prepare_for_class(__lowerCamelCase     , __lowerCamelCase			)			)
           SCREAMING_SNAKE_CASE										=   outputs.encoder_attentions if config.is_encoder_decoder else outputs.attentions
           self.assertEqual(len(__lowerCamelCase			)     , 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(__lowerCamelCase			)
           model.to(__lowerCamelCase			)
           model.eval()
           with torch.no_grad():
               SCREAMING_SNAKE_CASE										=   model(**self._prepare_for_class(__lowerCamelCase     , __lowerCamelCase			)			)
           SCREAMING_SNAKE_CASE										=   outputs.encoder_attentions
           self.assertEqual(len(__lowerCamelCase			)     , 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(__lowerCamelCase			)

           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(__lowerCamelCase     , __lowerCamelCase			)

           # decoder attentions
           SCREAMING_SNAKE_CASE										=   outputs.decoder_attentions
           self.assertIsInstance(__lowerCamelCase     , (list, tuple)			)
           self.assertEqual(len(__lowerCamelCase			)     , 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(__lowerCamelCase     , (list, tuple)			)
           self.assertEqual(len(__lowerCamelCase			)     , 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(__lowerCamelCase			)
       model.to(__lowerCamelCase			)
       model.eval()
       with torch.no_grad():
           SCREAMING_SNAKE_CASE										=   model(**self._prepare_for_class(__lowerCamelCase     , __lowerCamelCase			)			)

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

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

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





def 	__a     (      A__			:  Optional[int]="train-batch.pt"       ):
    SCREAMING_SNAKE_CASE										=   hf_hub_download(repo_id="hf-internal-testing/tourism-monthly-batch"			,  filename=A__			,  repo_type="dataset"       )
    SCREAMING_SNAKE_CASE										=   torch.load(A__			,  map_location=A__       )
    return batch





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







   '''simple docstring'''




   def      _snake_case		(       self					:			Optional[int]			):
       SCREAMING_SNAKE_CASE										=   AutoformerModel.from_pretrained("huggingface/autoformer-tourism-monthly"			).to(__lowerCamelCase			)
       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     , __lowerCamelCase			)

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





   def      _snake_case		(       self					:			List[Any]			):
       SCREAMING_SNAKE_CASE										=   AutoformerForPrediction.from_pretrained("huggingface/autoformer-tourism-monthly"			).to(__lowerCamelCase			)
       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     , __lowerCamelCase			)

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





   def      _snake_case		(       self					:			Optional[Any]			):
       SCREAMING_SNAKE_CASE										=   AutoformerForPrediction.from_pretrained("huggingface/autoformer-tourism-monthly"			).to(__lowerCamelCase			)
       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     , __lowerCamelCase			)

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


import logging
from dataclasses import dataclass, field
from pathlib import Path
from typing import Optional, Union

from .generation.configuration_utils import GenerationConfig
from .training_args import TrainingArguments
from .utils import add_start_docstrings


__A			:					Optional[Any]						=			logging.getLogger(__name__)





@dataclass
@add_start_docstrings(TrainingArguments.__doc__		)
class 			_SCREAMING_SNAKE_CASE (   __snake_case		):







   '''simple docstring'''




   lowerCamelCase__				=      field(default=__snake_case					,       metadata={"help": "Whether to use SortishSampler or not."}		)
   lowerCamelCase__				=      field(
       default=__snake_case					,       metadata={"help": "Whether to use generate to calculate generative metrics (ROUGE, BLEU)."}		)
   lowerCamelCase__				=      field(
       default=__snake_case					,       metadata={
           "help": (
               "The `max_length` to use on each evaluation loop when `predict_with_generate=True`. Will default "
               "to the `max_length` value of the model configuration."
           )
       }					,       )
   lowerCamelCase__				=      field(
       default=__snake_case					,       metadata={
           "help": (
               "The `num_beams` to use on each evaluation loop when `predict_with_generate=True`. Will default "
               "to the `num_beams` value of the model configuration."
           )
       }					,       )
   lowerCamelCase__				=      field(
       default=__snake_case					,       metadata={
           "help": "Model id, file path or url pointing to a GenerationConfig json file, to use during prediction."
       }					,       )





   def      _snake_case		(       self					:			Union[str, Any]			):
       SCREAMING_SNAKE_CASE										=   super().to_dict()
       for k, v in d.items():
           if isinstance(__lowerCamelCase     , __lowerCamelCase			):
               SCREAMING_SNAKE_CASE										=   v.to_dict()
       return d