


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


__magic_name__			=      1e-4

if is_torch_available():
     import torch

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



@require_torch
class 	__lowerCAmelCase		:


  '''simple docstring'''







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

        '''simple docstring'''




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

        A_						:  List[Any]          =				context_length
        A_						:  Union[str, Any]          =				prediction_length + label_length
        A_						:  int          =				label_length

        A_						:  Dict          =				moving_average
        A_						:  int          =				autocorrelation_factor






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

        '''simple docstring'''




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

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

        # decoder inputs
        A_						:  Tuple          =				floats_tensor([self.batch_size, config.prediction_length, config.num_time_features] )
        A_						:  Dict          =				floats_tensor([self.batch_size, config.prediction_length] )

        A_						:  Any          =				{
            """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 :       Dict ):

        '''simple docstring'''




        A_						:  List[Any]          =				self.get_config()
        A_						:  int          =				self.prepare_autoformer_inputs_dict(_a )
        return config, inputs_dict






  def        _a							(							self :       Dict ):

        '''simple docstring'''




        A_   ,					A_						:  List[str]          =				self.prepare_config_and_inputs()
        return config, inputs_dict






  def        _a							(							self :       Dict	,_a :       Dict	,_a :       Dict ):

        '''simple docstring'''




        A_						:  Optional[Any]          =				AutoformerModel(config=_a ).to(_a ).eval()
        A_						:  Optional[Any]          =				model(**_a )

        A_						:  Tuple          =				outputs.encoder_last_hidden_state
        A_						:  Any          =				outputs.last_hidden_state

        with tempfile.TemporaryDirectory() as tmpdirname:
              A_						:  Optional[Any]          =				model.get_encoder()
              encoder.save_pretrained(_a )
              A_						:  Tuple          =				AutoformerEncoder.from_pretrained(_a ).to(_a )

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

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

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

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

        A_						:  Any          =				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 	__lowerCAmelCase		(							__SCREAMING_SNAKE_CASE  , __SCREAMING_SNAKE_CASE  , unittest.TestCase    ):


  '''simple docstring'''







  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 :       str ):

        '''simple docstring'''




        A_						:  List[str]          =				AutoformerModelTester(self )
        A_						:  Tuple          =				ConfigTester(self	,config_class=_a	,has_text_modality=_a )






  def        _a							(							self :       Any ):

        '''simple docstring'''




        self.config_tester.run_common_tests()






  def        _a							(							self :       Dict ):

        '''simple docstring'''




        A_   ,					A_						:  Optional[int]          =				self.model_tester.prepare_config_and_inputs()
        for model_class in self.all_model_classes:
              A_						:  List[str]          =				model_class(_a )

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






  def        _a							(							self :       Any ):

        '''simple docstring'''




        A_						:  Optional[int]          =				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[str] ):

        '''simple docstring'''




        pass






  def        _a							(							self :       List[Any] ):

        '''simple docstring'''




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






  def        _a							(							self :       str ):

        '''simple docstring'''




        A_   ,					A_						:  str          =				self.model_tester.prepare_config_and_inputs_for_common()

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

              A_						:  str          =				[
                  """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 :       Dict ):

        '''simple docstring'''




        A_   ,					A_						:  int          =				self.model_tester.prepare_config_and_inputs_for_common()
        A_						:  int          =				True

        A_						:  str          =				getattr(self.model_tester	,"""seq_length"""	,_a )
        A_						:  Union[str, Any]          =				getattr(self.model_tester	,"""decoder_seq_length"""	,_a )
        A_						:  int          =				getattr(self.model_tester	,"""encoder_seq_length"""	,_a )
        A_						:  Union[str, Any]          =				getattr(self.model_tester	,"""d_model"""	,_a )
        A_						:  List[str]          =				getattr(self.model_tester	,"""num_attention_heads"""	,_a )
        A_						:  Tuple          =				d_model // num_attention_heads

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

              A_						:  Dict          =				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
              A_						:  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
              A_						:  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
        A_						:  int          =				True
        A_						:  Dict          =				True
        A_						:  Dict          =				model_class(_a )
        model.to(_a )
        model.eval()
        with torch.no_grad():
              A_						:  Any          =				model(**self._prepare_for_class(_a	,_a ) )

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

        A_						:  List[str]          =				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 :       str ):

        '''simple docstring'''




        super().test_retain_grad_hidden_states_attentions()







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






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


  '''simple docstring'''







  def        _a							(							self :       Any ):

        '''simple docstring'''




        A_						:  Optional[int]          =				AutoformerModel.from_pretrained("""huggingface/autoformer-tourism-monthly""" ).to(_a )
        A_						:  Optional[int]          =				prepare_batch()

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

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

        A_						:  Optional[Any]          =				torch.tensor(
            [[0.3593, -1.3398, 0.6330], [0.2279, 1.5396, -0.1792], [0.0450, 1.3225, -0.2335]]	,device=_a )
        self.assertTrue(torch.allclose(output[0, :3, :3]	,_a	,atol=_a ) )






  def        _a							(							self :       List[str] ):

        '''simple docstring'''




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

        A_						:  Union[str, Any]          =				torch.tensor(
            [[-0.0734, -0.9036, 0.8358], [4.7186, 2.4113, 1.9581], [1.7953, 2.3558, 1.2970]]	,device=_a )
        self.assertTrue(torch.allclose(output[0, :3, :3]	,_a	,atol=_a ) )






  def        _a							(							self :       Tuple ):

        '''simple docstring'''




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

        A_						:  int          =				torch.tensor([3130.6763, 4056.5293, 7053.0786]	,device=_a )
        A_						:  Tuple          =				outputs.sequences.mean(dim=1 )
        self.assertTrue(torch.allclose(mean_prediction[0, -3:]	,_a	,rtol=1e-1 ) )




'''simple docstring'''


from ..utils import DummyObject, requires_backends



class 	__lowerCAmelCase		(							metaclass=__SCREAMING_SNAKE_CASE    ):


  '''simple docstring'''







  a_  					=							["""torch""", """torchsde"""]






  def __init__(							self :       Any	,*_a :       Union[str, Any]	,**_a :       Optional[int] ):

        '''simple docstring'''




        requires_backends(self	,["""torch""", """torchsde"""] )






  @classmethod
  def        _a							(							cls :       Optional[int]	,*_a :       List[Any]	,**_a :       Any ):

        '''simple docstring'''




        requires_backends(cls	,["""torch""", """torchsde"""] )






  @classmethod
  def        _a							(							cls :       List[Any]	,*_a :       Tuple	,**_a :       Union[str, Any] ):

        '''simple docstring'''




        requires_backends(cls	,["""torch""", """torchsde"""] )
