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


__lowercase		:		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	:




    """simple docstring"""




    def __init__(		self							:     str	,      A_							:     Optional[int]	,      A_							:     Union[str, Any]=16	,      A_							:     int=13	,      A_							:     Dict=7	,      A_							:     Optional[int]=14	,      A_							:     str=10	,      A_							:     str=19	,      A_							:     int=5	,      A_							:     int=4	,      A_							:     Optional[Any]=True	,      A_							:     List[str]=16	,      A_							:     Union[str, Any]=2	,      A_							:     Dict=4	,      A_							:     Tuple=4	,      A_							:     Any="gelu"	,      A_							:     List[Any]=0.1	,      A_							:     Any=0.1	,      A_							:     List[str]=[1, 2, 3, 4, 5]	,      A_							:     Optional[int]=25	,      A_							:     Optional[Any]=5	,      )    ->       List[Any]:
         __snake_case				=							d_model
         __snake_case				=							parent
         __snake_case				=							batch_size
         __snake_case				=							prediction_length
         __snake_case				=							context_length
         __snake_case				=							cardinality
         __snake_case				=							num_time_features
         __snake_case				=							lags_sequence
         __snake_case				=							embedding_dimension
         __snake_case				=							is_training
         __snake_case				=							hidden_size
         __snake_case				=							num_hidden_layers
         __snake_case				=							num_attention_heads
         __snake_case				=							intermediate_size
         __snake_case				=							hidden_act
         __snake_case				=							hidden_dropout_prob
         __snake_case				=							attention_probs_dropout_prob

         __snake_case				=							context_length
         __snake_case				=							prediction_length + label_length
         __snake_case				=							label_length

         __snake_case				=							moving_average
         __snake_case				=							autocorrelation_factor
    def  lowercase					(		self							:     Tuple    )    ->       Optional[int]:
         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  lowercase					(		self							:     List[Any]	,      A_							:     Dict    )    ->       Union[str, Any]:
         __snake_case				=							config.context_length + max(config.lags_sequence    )

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

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

         __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  lowercase					(		self							:     Tuple    )    ->       Optional[Any]:
         __snake_case				=							self.get_config()
         __snake_case				=							self.prepare_autoformer_inputs_dict(a_    )
         return config, inputs_dict
    def  lowercase					(		self							:     Dict    )    ->       Optional[Any]:
         __snake_case				=							self.prepare_config_and_inputs()
         return config, inputs_dict







    def  lowercase					(		self							:     Union[str, Any]	,      A_							:     Union[str, Any]	,      A_							:     List[str]    )    ->       Optional[int]:
         __snake_case				=							AutoformerModel(config=a_    ).to(a_    ).eval()
         __snake_case				=							model(**a_    )

         __snake_case				=							outputs.encoder_last_hidden_state
         __snake_case				=							outputs.last_hidden_state

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

         __snake_case				=							model.create_network_inputs(**a_    )
         __snake_case				=							model.decomposition_layer(transformer_inputs[:, : config.context_length, ...]    )

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

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

         __snake_case				=							torch.cat(
             (
                 torch.cat((seasonal_input[:, -config.label_length :, ...], zeros)	,      dim=1    ),
                 feature[:, config.context_length - config.label_length :, ...],
             )	,      dim=-1	,      )
         __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:
              __snake_case				=							model.get_decoder()
              decoder.save_pretrained(a_    )
              __snake_case				=							AutoformerDecoder.from_pretrained(a_    ).to(a_    )

         __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	(						__lowerCamelCase							, __lowerCamelCase							, unittest.TestCase					):




    """simple docstring"""




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

              with tempfile.TemporaryDirectory() as tmpdirname:
                   model.save_pretrained(a_    )
                   __snake_case				=							model_class.from_pretrained(a_	,      output_loading_info=a_    )
              self.assertEqual(info['''missing_keys''']	,      []    )
    def  lowercase					(		self							:     List[str]    )    ->       Optional[int]:
         __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  lowercase					(		self							:     Any    )    ->       int:
         pass
    def  lowercase					(		self							:     str    )    ->       List[str]:
         __snake_case				=							inspect.signature(getattr(a_	,      '''forward'''    )    )
         # The main input is the name of the argument after `self`
         __snake_case				=							list(model_signature.parameters.keys()    )[1]
         self.assertEqual(AutoformerModel.main_input_name	,      a_    )
    def  lowercase					(		self							:     Optional[int]    )    ->       Union[str, Any]:
         __snake_case				=							self.model_tester.prepare_config_and_inputs_for_common()

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

              __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  lowercase					(		self							:     Any    )    ->       List[Any]:
         __snake_case				=							self.model_tester.prepare_config_and_inputs_for_common()
         __snake_case				=							True

         __snake_case				=							getattr(self.model_tester	,      '''seq_length'''	,      a_    )
         __snake_case				=							getattr(self.model_tester	,      '''decoder_seq_length'''	,      a_    )
         __snake_case				=							getattr(self.model_tester	,      '''encoder_seq_length'''	,      a_    )
         __snake_case				=							getattr(self.model_tester	,      '''d_model'''	,      a_    )
         __snake_case				=							getattr(self.model_tester	,      '''num_attention_heads'''	,      a_    )
         __snake_case				=							d_model // num_attention_heads

         for model_class in self.all_model_classes:
              __snake_case				=							True
              __snake_case				=							False
              __snake_case				=							True
              __snake_case				=							model_class(a_    )
              model.to(a_    )
              model.eval()
              with torch.no_grad():
                   __snake_case				=							model(**self._prepare_for_class(a_	,      a_    )    )
              __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"]
              __snake_case				=							True
              __snake_case				=							model_class(a_    )
              model.to(a_    )
              model.eval()
              with torch.no_grad():
                   __snake_case				=							model(**self._prepare_for_class(a_	,      a_    )    )
              __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]	,      )
              __snake_case				=							len(a_    )

              __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
              __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
              __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
         __snake_case				=							True
         __snake_case				=							True
         __snake_case				=							model_class(a_    )
         model.to(a_    )
         model.eval()
         with torch.no_grad():
              __snake_case				=							model(**self._prepare_for_class(a_	,      a_    )    )

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

         __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  lowercase					(		self							:     int    )    ->       Dict:
         super().test_retain_grad_hidden_states_attentions()






def    SCREAMING_SNAKE_CASE       (     snake_case="train-batch.pt"):
     __snake_case				=							hf_hub_download(repo_id='''hf-internal-testing/tourism-monthly-batch''',      filename=lowerCAmelCase_,      repo_type='''dataset''')
     __snake_case				=							torch.load(lowerCAmelCase_,      map_location=lowerCAmelCase_)
     return batch





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




    """simple docstring"""




    def  lowercase					(		self							:     Union[str, Any]    )    ->       Tuple:
         __snake_case				=							AutoformerModel.from_pretrained('''huggingface/autoformer-tourism-monthly'''    ).to(a_    )
         __snake_case				=							prepare_batch()

         with torch.no_grad():
              __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]

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

         __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  lowercase					(		self							:     Dict    )    ->       Optional[int]:
         __snake_case				=							AutoformerForPrediction.from_pretrained('''huggingface/autoformer-tourism-monthly'''    ).to(a_    )
         __snake_case				=							prepare_batch('''val-batch.pt'''    )
         with torch.no_grad():
              __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
         __snake_case				=							torch.Size((64, model.config.context_length, model.config.d_model)    )
         self.assertEqual(output.shape	,      a_    )

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

         __snake_case				=							torch.tensor([3_130.6_763, 4_056.5_293, 7_053.0_786]	,      device=a_    )
         __snake_case				=							outputs.sequences.mean(dim=1    )
         self.assertTrue(torch.allclose(mean_prediction[0, -3:]	,      a_	,      rtol=1E-1    )    )
"""simple docstring"""



from __future__ import annotations

from collections.abc import Iterator
from typing import Any



class      _A	:




    """simple docstring"""




    def __init__(		self							:     Tuple	,      A_							:     Any    )    ->       Union[str, Any]:
         __snake_case				=							data
         __snake_case				=							None






class      _A	:




    """simple docstring"""




    def __init__(		self							:     Union[str, Any]    )    ->       List[Any]:
         __snake_case				=							None
         __snake_case				=							None
    def __iter__(		self							:     Optional[Any]    )    ->       Iterator[Any]:
         __snake_case				=							self.head
         while self.head:
              yield node.data
              __snake_case				=							node.next
              if node == self.head:
                   break
    def __len__(		self							:     List[Any]    )    ->       int:
         return sum(1 for _ in self    )
    def __repr__(		self							:     Optional[int]    )    ->       Union[str, Any]:
         return "->".join(str(A_    ) for item in iter(self    )    )
    def  lowercase					(		self							:     int	,      A_							:     Any    )    ->       None:
         self.insert_nth(len(self    )	,      A_    )
    def  lowercase					(		self							:     List[str]	,      A_							:     Any    )    ->       None:
         self.insert_nth(0	,      A_    )
    def  lowercase					(		self							:     List[Any]	,      A_							:     int	,      A_							:     Any    )    ->       None:
         if index < 0 or index > len(self    ):
              raise IndexError('''list index out of range.'''    )
         __snake_case				=							Node(A_    )
         if self.head is None:
              __snake_case				=							new_node  # first node points itself
              __snake_case				=							__snake_case				=							new_node
         elif index == 0:  # insert at head
              __snake_case				=							self.head
              __snake_case				=							__snake_case				=							new_node
         else:
              __snake_case				=							self.head
              for _ in range(index - 1    ):
                   __snake_case				=							temp.next
              __snake_case				=							temp.next
              __snake_case				=							new_node
              if index == len(self    ) - 1:  # insert at tail
                   __snake_case				=							new_node
    def  lowercase					(		self							:     int    )    ->       str:
         return self.delete_nth(0    )
    def  lowercase					(		self							:     List[Any]    )    ->       Any:
         return self.delete_nth(len(self    ) - 1    )
    def  lowercase					(		self							:     List[str]	,      A_							:     int = 0    )    ->       Any:
         if not 0 <= index < len(self    ):
              raise IndexError('''list index out of range.'''    )
         __snake_case				=							self.head
         if self.head == self.tail:  # just one node
              __snake_case				=							__snake_case				=							None
         elif index == 0:  # delete head node
              __snake_case				=							self.tail.next.next
              __snake_case				=							self.head.next
         else:
              __snake_case				=							self.head
              for _ in range(index - 1    ):
                   __snake_case				=							temp.next
              __snake_case				=							temp.next
              __snake_case				=							temp.next.next
              if index == len(self    ) - 1:  # delete at tail
                   __snake_case				=							temp
         return delete_node.data







    def  lowercase					(		self							:     Dict    )    ->       bool:
         return len(self    ) == 0






def    SCREAMING_SNAKE_CASE       (     ):
     __snake_case				=							CircularLinkedList()
     assert len(snake_case) == 0
     assert circular_linked_list.is_empty() is True
     assert str(snake_case) == ""

     try:
          circular_linked_list.delete_front()
          raise AssertionError  # This should not happen
     except IndexError:
          assert True  # This should happen

     try:
          circular_linked_list.delete_tail()
          raise AssertionError  # This should not happen
     except IndexError:
          assert True  # This should happen

     try:
          circular_linked_list.delete_nth(-1)
          raise AssertionError
     except IndexError:
          assert True

     try:
          circular_linked_list.delete_nth(0)
          raise AssertionError
     except IndexError:
          assert True

     assert circular_linked_list.is_empty() is True
     for i in range(5):
          assert len(snake_case) == i
          circular_linked_list.insert_nth(snake_case,      i + 1)
     assert str(snake_case) == "->".join(str(snake_case) for i in range(1,      6))

     circular_linked_list.insert_tail(6)
     assert str(snake_case) == "->".join(str(snake_case) for i in range(1,      7))
     circular_linked_list.insert_head(0)
     assert str(snake_case) == "->".join(str(snake_case) for i in range(0,      7))

     assert circular_linked_list.delete_front() == 0
     assert circular_linked_list.delete_tail() == 6
     assert str(snake_case) == "->".join(str(snake_case) for i in range(1,      6))
     assert circular_linked_list.delete_nth(2) == 3

     circular_linked_list.insert_nth(2,      3)
     assert str(snake_case) == "->".join(str(snake_case) for i in range(1,      6))

     assert circular_linked_list.is_empty() is False


if __name__ == "__main__":
       import doctest

       doctest.testmod()