






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		:       Dict			     = 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					:			Any		,      __lowercase					:			int		,      __lowercase					:			List[Any]=16		,      __lowercase					:			Union[str, Any]=13		,      __lowercase					:			int=7		,      __lowercase					:			int=14		,      __lowercase					:			Union[str, Any]=10		,      __lowercase					:			int=19		,      __lowercase					:			Any=5		,      __lowercase					:			List[str]=4		,      __lowercase					:			str=True		,      __lowercase					:			Union[str, Any]=16		,      __lowercase					:			str=2		,      __lowercase					:			Any=4		,      __lowercase					:			Tuple=4		,      __lowercase					:			str="gelu"		,      __lowercase					:			Optional[Any]=0.1		,      __lowercase					:			List[str]=0.1		,      __lowercase					:			Any=[1, 2, 3, 4, 5]		,      __lowercase					:			Dict=25		,      __lowercase					:			Dict=5		,      )							->      str:
        __UpperCAmelCase							:	Optional[Any]           =  d_model
        __UpperCAmelCase							:	List[Any]           =  parent
        __UpperCAmelCase							:	Union[str, Any]           =  batch_size
        __UpperCAmelCase							:	Dict           =  prediction_length
        __UpperCAmelCase							:	Optional[int]           =  context_length
        __UpperCAmelCase							:	int           =  cardinality
        __UpperCAmelCase							:	Optional[Any]           =  num_time_features
        __UpperCAmelCase							:	Any           =  lags_sequence
        __UpperCAmelCase							:	Union[str, Any]           =  embedding_dimension
        __UpperCAmelCase							:	List[str]           =  is_training
        __UpperCAmelCase							:	List[Any]           =  hidden_size
        __UpperCAmelCase							:	Optional[int]           =  num_hidden_layers
        __UpperCAmelCase							:	Optional[Any]           =  num_attention_heads
        __UpperCAmelCase							:	Dict           =  intermediate_size
        __UpperCAmelCase							:	Tuple           =  hidden_act
        __UpperCAmelCase							:	List[str]           =  hidden_dropout_prob
        __UpperCAmelCase							:	List[Any]           =  attention_probs_dropout_prob

        __UpperCAmelCase							:	Any           =  context_length
        __UpperCAmelCase							:	Union[str, Any]           =  prediction_length + label_length
        __UpperCAmelCase							:	Dict           =  label_length

        __UpperCAmelCase							:	Any           =  moving_average
        __UpperCAmelCase							:	Optional[int]           =  autocorrelation_factor
      def 	UpperCAmelCase      (      self					:			List[Any]					)							->      str:
        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					:			Optional[int]		,      __lowercase					:			Optional[int]					)							->      List[str]:
        __UpperCAmelCase							:	List[Any]           =  config.context_length + max(config.lags_sequence					)

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

        # decoder inputs
        __UpperCAmelCase							:	List[Any]           =  floats_tensor([self.batch_size, config.prediction_length, config.num_time_features]					)
        __UpperCAmelCase							:	List[Any]           =  floats_tensor([self.batch_size, config.prediction_length]					)

        __UpperCAmelCase							:	List[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 	UpperCAmelCase      (      self					:			Optional[Any]					)							->      Dict:
        __UpperCAmelCase							:	Optional[int]           =  self.get_config()
        __UpperCAmelCase							:	Union[str, Any]           =  self.prepare_autoformer_inputs_dict(__lowercase					)
        return config, inputs_dict
      def 	UpperCAmelCase      (      self					:			Union[str, Any]					)							->      int:
        __UpperCAmelCase      , __UpperCAmelCase							:	Optional[int]           =  self.prepare_config_and_inputs()
        return config, inputs_dict




      def 	UpperCAmelCase      (      self					:			Any		,      __lowercase					:			Optional[int]		,      __lowercase					:			Optional[int]					)							->      Dict:
        __UpperCAmelCase							:	Any           =  AutoformerModel(config=__lowercase					).to(__lowercase					).eval()
        __UpperCAmelCase							:	Any           =  model(**__lowercase					)

        __UpperCAmelCase							:	Tuple           =  outputs.encoder_last_hidden_state
        __UpperCAmelCase							:	Optional[int]           =  outputs.last_hidden_state

        with tempfile.TemporaryDirectory() as tmpdirname:
          __UpperCAmelCase							:	str           =  model.get_encoder()
          encoder.save_pretrained(__lowercase					)
          __UpperCAmelCase							:	Dict           =  AutoformerEncoder.from_pretrained(__lowercase					).to(__lowercase					)

        __UpperCAmelCase      , __UpperCAmelCase      , __UpperCAmelCase      , __UpperCAmelCase      , __UpperCAmelCase							:	str           =  model.create_network_inputs(**__lowercase					)
        __UpperCAmelCase      , __UpperCAmelCase							:	Optional[int]           =  model.decomposition_layer(transformer_inputs[:, : config.context_length, ...]					)

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

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

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

        __UpperCAmelCase							:	Optional[int]           =  decoder(
            trend=__lowercase		,      inputs_embeds=__lowercase		,      encoder_hidden_states=__lowercase		,      )[0]

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


@require_torch
class     a    ( lowercase__							,       lowercase__							,       unittest.TestCase   ):



      """simple docstring"""


      a  : int							       =	(AutoformerModel, AutoformerForPrediction) if is_torch_available() else ()
      a  : Optional[Any]							       =	(AutoformerForPrediction,) if is_torch_available() else ()
      a  : Union[str, Any]							       =	{'feature-extraction': AutoformerModel} if is_torch_available() else {}
      a  : int							       =	False
      a  : List[Any]							       =	False
      a  : Tuple							       =	False
      a  : Optional[int]							       =	False
      a  : Any							       =	False
      a  : Dict							       =	False
      def 	UpperCAmelCase      (      self					:			Dict					)							->      Dict:
        __UpperCAmelCase							:	Tuple           =  AutoformerModelTester(self					)
        __UpperCAmelCase							:	Any           =  ConfigTester(self		,      config_class=__lowercase		,      has_text_modality=__lowercase					)
      def 	UpperCAmelCase      (      self					:			List[str]					)							->      str:
        self.config_tester.run_common_tests()
      def 	UpperCAmelCase      (      self					:			Dict					)							->      Optional[Any]:
        __UpperCAmelCase      , __UpperCAmelCase							:	Dict           =  self.model_tester.prepare_config_and_inputs()
        for model_class in self.all_model_classes:
          __UpperCAmelCase							:	Dict           =  model_class(__lowercase					)

          with tempfile.TemporaryDirectory() as tmpdirname:
            model.save_pretrained(__lowercase					)
            __UpperCAmelCase      , __UpperCAmelCase							:	Union[str, Any]           =  model_class.from_pretrained(__lowercase		,      output_loading_info=__lowercase					)
          self.assertEqual(info["""missing_keys"""]		,      []					)
      def 	UpperCAmelCase      (      self					:			Tuple					)							->      int:
        __UpperCAmelCase							:	Tuple           =  self.model_tester.prepare_config_and_inputs_for_common()
        self.model_tester.check_encoder_decoder_model_standalone(*__lowercase					)
      @unittest.skip(reason="""Model has no tokens embeddings"""					)
      def 	UpperCAmelCase      (      self					:			int					)							->      Any:
        pass
      def 	UpperCAmelCase      (      self					:			Union[str, Any]					)							->      List[str]:
        __UpperCAmelCase							:	Any           =  inspect.signature(getattr(__lowercase		,      """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		,      __lowercase					)
      def 	UpperCAmelCase      (      self					:			Optional[int]					)							->      Optional[int]:
        __UpperCAmelCase      , __UpperCAmelCase							:	Union[str, Any]           =  self.model_tester.prepare_config_and_inputs_for_common()

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

          __UpperCAmelCase							:	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(__lowercase					)]		,      __lowercase					)
      def 	UpperCAmelCase      (      self					:			Union[str, Any]					)							->      Tuple:
        __UpperCAmelCase      , __UpperCAmelCase							:	List[str]           =  self.model_tester.prepare_config_and_inputs_for_common()
        __UpperCAmelCase							:	List[str]           =  True

        __UpperCAmelCase							:	str           =  getattr(self.model_tester		,      """seq_length"""		,      __lowercase					)
        __UpperCAmelCase							:	Dict           =  getattr(self.model_tester		,      """decoder_seq_length"""		,      __lowercase					)
        __UpperCAmelCase							:	Dict           =  getattr(self.model_tester		,      """encoder_seq_length"""		,      __lowercase					)
        __UpperCAmelCase							:	int           =  getattr(self.model_tester		,      """d_model"""		,      __lowercase					)
        __UpperCAmelCase							:	List[str]           =  getattr(self.model_tester		,      """num_attention_heads"""		,      __lowercase					)
        __UpperCAmelCase							:	int           =  d_model // num_attention_heads

        for model_class in self.all_model_classes:
          __UpperCAmelCase							:	Dict           =  True
          __UpperCAmelCase							:	Dict           =  False
          __UpperCAmelCase							:	Any           =  True
          __UpperCAmelCase							:	Tuple           =  model_class(__lowercase					)
          model.to(__lowercase					)
          model.eval()
          with torch.no_grad():
            __UpperCAmelCase							:	Optional[int]           =  model(**self._prepare_for_class(__lowercase		,      __lowercase					)					)
          __UpperCAmelCase							:	Tuple           =  outputs.encoder_attentions if config.is_encoder_decoder else outputs.attentions
          self.assertEqual(len(__lowercase					)		,      self.model_tester.num_hidden_layers					)

          # check that output_attentions also work using config
          del inputs_dict["output_attentions"]
          __UpperCAmelCase							:	Union[str, Any]           =  True
          __UpperCAmelCase							:	List[Any]           =  model_class(__lowercase					)
          model.to(__lowercase					)
          model.eval()
          with torch.no_grad():
            __UpperCAmelCase							:	List[Any]           =  model(**self._prepare_for_class(__lowercase		,      __lowercase					)					)
          __UpperCAmelCase							:	Optional[int]           =  outputs.encoder_attentions
          self.assertEqual(len(__lowercase					)		,      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[int]           =  len(__lowercase					)

          __UpperCAmelCase							:	Optional[Any]           =  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(__lowercase		,      __lowercase					)

          # decoder attentions
          __UpperCAmelCase							:	Any           =  outputs.decoder_attentions
          self.assertIsInstance(__lowercase		,      (list, tuple)					)
          self.assertEqual(len(__lowercase					)		,      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							:	str           =  outputs.cross_attentions
          self.assertIsInstance(__lowercase		,      (list, tuple)					)
          self.assertEqual(len(__lowercase					)		,      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							:	Optional[int]           =  True
        __UpperCAmelCase							:	Tuple           =  True
        __UpperCAmelCase							:	Optional[Any]           =  model_class(__lowercase					)
        model.to(__lowercase					)
        model.eval()
        with torch.no_grad():
          __UpperCAmelCase							:	str           =  model(**self._prepare_for_class(__lowercase		,      __lowercase					)					)

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

        __UpperCAmelCase							:	int           =  outputs.encoder_attentions if config.is_encoder_decoder else outputs.attentions

        self.assertEqual(len(__lowercase					)		,      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					:			Optional[int]					)							->      Tuple:
        super().test_retain_grad_hidden_states_attentions()






def 				lowerCamelCase__      (    __lowerCamelCase			:     Any="train-batch.pt"		):
  __UpperCAmelCase							:	Tuple           =  hf_hub_download(repo_id="""hf-internal-testing/tourism-monthly-batch"""  ,		filename=__lowerCamelCase  ,		repo_type="""dataset"""		)
  __UpperCAmelCase							:	str           =  torch.load(__lowerCamelCase  ,		map_location=__lowerCamelCase		)
  return batch







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



      """simple docstring"""


      def 	UpperCAmelCase      (      self					:			Optional[Any]					)							->      int:
        __UpperCAmelCase							:	Union[str, Any]           =  AutoformerModel.from_pretrained("""huggingface/autoformer-tourism-monthly"""					).to(__lowercase					)
        __UpperCAmelCase							:	str           =  prepare_batch()

        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"""]		,      future_values=batch["""future_values"""]		,      future_time_features=batch["""future_time_features"""]		,      )[0]

        __UpperCAmelCase							:	Any           =  torch.Size(
            (64, model.config.prediction_length + model.config.label_length, model.config.feature_size)					)
        self.assertEqual(output.shape		,      __lowercase					)

        __UpperCAmelCase							:	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=__lowercase					)
        self.assertTrue(torch.allclose(output[0, :3, :3]		,      __lowercase		,      atol=__lowercase					)					)
      def 	UpperCAmelCase      (      self					:			int					)							->      int:
        __UpperCAmelCase							:	int           =  AutoformerForPrediction.from_pretrained("""huggingface/autoformer-tourism-monthly"""					).to(__lowercase					)
        __UpperCAmelCase							:	str           =  prepare_batch("""val-batch.pt"""					)
        with torch.no_grad():
          __UpperCAmelCase							:	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
        __UpperCAmelCase							:	List[Any]           =  torch.Size((64, model.config.context_length, model.config.d_model)					)
        self.assertEqual(output.shape		,      __lowercase					)

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




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

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










from abc import ABC, abstractmethod
from typing import Optional, Union

from .. import Dataset, DatasetDict, Features, IterableDataset, IterableDatasetDict, NamedSplit
from ..utils.typing import NestedDataStructureLike, PathLike
class     a    ( lowercase__   ):



      """simple docstring"""


      def __init__(      self					:			Any		,      __lowercase					:			Optional[NestedDataStructureLike[PathLike]] = None		,      __lowercase					:			Optional[NamedSplit] = None		,      __lowercase					:			Optional[Features] = None		,      __lowercase					:			str = None		,      __lowercase					:			bool = False		,      __lowercase					:			bool = False		,      __lowercase					:			Optional[int] = None		,      **__lowercase					:			List[str]		,      )							->      Tuple:
        __UpperCAmelCase							:	Any           =  path_or_paths
        __UpperCAmelCase							:	Dict           =  split if split or isinstance(__lowercase		,      __lowercase					) else """train"""
        __UpperCAmelCase							:	Optional[int]           =  features
        __UpperCAmelCase							:	str           =  cache_dir
        __UpperCAmelCase							:	str           =  keep_in_memory
        __UpperCAmelCase							:	Optional[int]           =  streaming
        __UpperCAmelCase							:	Dict           =  num_proc
        __UpperCAmelCase							:	Tuple           =  kwargs




      @abstractmethod
      def 	UpperCAmelCase      (      self					:			List[Any]					)							->      Union[Dataset, DatasetDict, IterableDataset, IterableDatasetDict]:
        pass


class     a    ( lowercase__   ):



      """simple docstring"""


      def __init__(      self					:			List[str]		,      __lowercase					:			Optional[Features] = None		,      __lowercase					:			str = None		,      __lowercase					:			bool = False		,      __lowercase					:			bool = False		,      __lowercase					:			Optional[int] = None		,      **__lowercase					:			Optional[Any]		,      )							->      Optional[int]:
        __UpperCAmelCase							:	Optional[Any]           =  features
        __UpperCAmelCase							:	str           =  cache_dir
        __UpperCAmelCase							:	Optional[int]           =  keep_in_memory
        __UpperCAmelCase							:	Dict           =  streaming
        __UpperCAmelCase							:	Optional[Any]           =  num_proc
        __UpperCAmelCase							:	Union[str, Any]           =  kwargs




      @abstractmethod
      def 	UpperCAmelCase      (      self					:			List[str]					)							->      Union[Dataset, IterableDataset]:
        pass


