






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


UpperCamelCase							       =    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    lowercase_    :






      def __init__(    self     ,     a_     ,     a_=1_6     ,     a_=1_3     ,     a_=7     ,     a_=1_4     ,     a_=1_0     ,     a_=1_9     ,     a_=5     ,     a_=4     ,     a_=True     ,     a_=1_6     ,     a_=2     ,     a_=4     ,     a_=4     ,     a_="gelu"     ,     a_=0.1     ,     a_=0.1     ,     a_=[1, 2, 3, 4, 5]     ,     a_=2_5     ,     a_=5     ,     )	->List[str]:







        '''simple docstring'''


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

        _a							=	context_length
        _a							=	prediction_length + label_length
        _a							=	label_length

        _a							=	moving_average
        _a							=	autocorrelation_factor






      def 			lowerCamelCase__    (    self	)	->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 			lowerCamelCase__    (    self     ,     a_	)	->str:







        '''simple docstring'''


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

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

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

        _a							=	{
            "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 			lowerCamelCase__    (    self	)	->Any:







        '''simple docstring'''


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






      def 			lowerCamelCase__    (    self	)	->Dict:







        '''simple docstring'''


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






      def 			lowerCamelCase__    (    self     ,     a_     ,     a_	)	->Any:







        '''simple docstring'''


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

        _a							=	outputs.encoder_last_hidden_state
        _a							=	outputs.last_hidden_state

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

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

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

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

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

        _a							=	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    lowercase_    (_UpperCAmelCase,    _UpperCAmelCase,    unittest.TestCase     ):
      A__	:		Dict       			=				(AutoformerModel, AutoformerForPrediction) if is_torch_available() else ()
      A__	:		Tuple       			=				(AutoformerForPrediction,) if is_torch_available() else ()
      A__	:		List[Any]       			=				{'''feature-extraction''': AutoformerModel} if is_torch_available() else {}
      A__	:		Any       			=				False
      A__	:		List[Any]       			=				False
      A__	:		Optional[int]       			=				False
      A__	:		List[str]       			=				False
      A__	:		str       			=				False
      A__	:		List[Any]       			=				False






      def 			lowerCamelCase__    (    self	)	->List[str]:







        '''simple docstring'''


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






      def 			lowerCamelCase__    (    self	)	->List[Any]:







        '''simple docstring'''


        self.config_tester.run_common_tests()






      def 			lowerCamelCase__    (    self	)	->Optional[Any]:







        '''simple docstring'''


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

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






      def 			lowerCamelCase__    (    self	)	->Any:







        '''simple docstring'''


        _a							=	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 			lowerCamelCase__    (    self	)	->List[str]:







        '''simple docstring'''


        pass






      def 			lowerCamelCase__    (    self	)	->Dict:







        '''simple docstring'''


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






      def 			lowerCamelCase__    (    self	)	->int:







        '''simple docstring'''


        _a			,	_a							=	self.model_tester.prepare_config_and_inputs_for_common()

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

          _a							=	[
              "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 			lowerCamelCase__    (    self	)	->Dict:







        '''simple docstring'''


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

        _a							=	getattr(self.model_tester     ,     "seq_length"     ,     a_	)
        _a							=	getattr(self.model_tester     ,     "decoder_seq_length"     ,     a_	)
        _a							=	getattr(self.model_tester     ,     "encoder_seq_length"     ,     a_	)
        _a							=	getattr(self.model_tester     ,     "d_model"     ,     a_	)
        _a							=	getattr(self.model_tester     ,     "num_attention_heads"     ,     a_	)
        _a							=	d_model // num_attention_heads

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

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

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

        _a							=	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 			lowerCamelCase__    (    self	)	->int:







        '''simple docstring'''


        super().test_retain_grad_hidden_states_attentions()







def        lowerCAmelCase (      UpperCamelCase_:							int="train-batch.pt"					)       ->							int:







  '''simple docstring'''







  _a							=	hf_hub_download(repo_id="hf-internal-testing/tourism-monthly-batch"     , filename=UpperCamelCase_     , repo_type="dataset"					)
  _a							=	torch.load(UpperCamelCase_     , map_location=UpperCamelCase_					)
  return batch




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






      def 			lowerCamelCase__    (    self	)	->str:







        '''simple docstring'''


        _a							=	AutoformerModel.from_pretrained("huggingface/autoformer-tourism-monthly"	).to(a_	)
        _a							=	prepare_batch()

        with torch.no_grad():
          _a							=	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							=	torch.Size(
            (6_4, model.config.prediction_length + model.config.label_length, model.config.feature_size)	)
        self.assertEqual(output.shape     ,     a_	)

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






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







        '''simple docstring'''


        _a							=	AutoformerForPrediction.from_pretrained("huggingface/autoformer-tourism-monthly"	).to(a_	)
        _a							=	prepare_batch("val-batch.pt"	)
        with torch.no_grad():
          _a							=	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							=	torch.Size((6_4, model.config.context_length, model.config.d_model)	)
        self.assertEqual(output.shape     ,     a_	)

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






      def 			lowerCamelCase__    (    self	)	->Any:







        '''simple docstring'''


        _a							=	AutoformerForPrediction.from_pretrained("huggingface/autoformer-tourism-monthly"	).to(a_	)
        _a							=	prepare_batch("val-batch.pt"	)
        with torch.no_grad():
          _a							=	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							=	torch.Size((6_4, model.config.num_parallel_samples, model.config.prediction_length)	)
        self.assertEqual(outputs.sequences.shape     ,     a_	)

        _a							=	torch.tensor([3_1_3_0.6_7_6_3, 4_0_5_6.5_2_9_3, 7_0_5_3.0_7_8_6]     ,     device=a_	)
        _a							=	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 Callable
from typing import Generic, TypeVar

UpperCamelCase							       =    TypeVar("""T""")
UpperCamelCase							       =    TypeVar("""U""")




class    lowercase_    (Generic[T, U]     ):






      def __init__(    self     ,     a_     ,     a_	)	->Any:







        '''simple docstring'''


        _a							=	key
        _a							=	val
        _a							=	None
        _a							=	None






      def __repr__(    self	)	->str:







        '''simple docstring'''


        return (
            f'''Node: key: {self.key}, val: {self.val}, '''
            f'''has next: {bool(self.next	)}, has prev: {bool(self.prev	)}'''
        )







class    lowercase_    (Generic[T, U]     ):






      def __init__(    self	)	->None:







        '''simple docstring'''


        _a							=	DoubleLinkedListNode(a_     ,     a_	)
        _a							=	DoubleLinkedListNode(a_     ,     a_	)
        _a			,	_a							=	self.rear, self.head






      def __repr__(    self	)	->str:







        '''simple docstring'''


        _a							=	["DoubleLinkedList"]
        _a							=	self.head
        while node.next is not None:
          rep.append(str(a_	)	)
          _a							=	node.next
        rep.append(str(self.rear	)	)
        return ",\n    ".join(a_	)






      def 			lowerCamelCase__    (    self     ,     a_	)	->None:







        '''simple docstring'''


        _a							=	self.rear.prev

        # All nodes other than self.head are guaranteed to have non-None previous
        assert previous is not None

        _a							=	node
        _a							=	previous
        _a							=	node
        _a							=	self.rear






      def 			lowerCamelCase__    (    self     ,     a_	)	->DoubleLinkedListNode[T, U] | None:







        '''simple docstring'''


        if node.prev is None or node.next is None:
          return None

        _a							=	node.next
        _a							=	node.prev
        _a							=	None
        _a							=	None
        return node







class    lowercase_    (Generic[T, U]     ):
      A__	:		dict[Callable[[T], U], LRUCache[T, U]]       			=				{}






      def __init__(    self     ,     a_	)	->int:







        '''simple docstring'''


        _a							=	DoubleLinkedList()
        _a							=	capacity
        _a							=	0
        _a							=	0
        _a							=	0
        _a							=	{}






      def __repr__(    self	)	->str:







        '''simple docstring'''


        return (
            f'''CacheInfo(hits={self.hits}, misses={self.miss}, '''
            f'''capacity={self.capacity}, current size={self.num_keys})'''
        )






      def __contains__(    self     ,     a_	)	->bool:







        '''simple docstring'''


        return key in self.cache






      def 			lowerCamelCase__    (    self     ,     a_	)	->U | None:







        '''simple docstring'''


        if key in self.cache:
          self.hits += 1
          _a							=	self.cache[key]
          _a							=	self.list.remove(self.cache[key]	)
          assert node == value_node

          # node is guaranteed not None because it is in self.cache
          assert node is not None
          self.list.add(a_	)
          return node.val
        self.miss += 1
        return None






      def 			lowerCamelCase__    (    self     ,     a_     ,     a_	)	->None:







        '''simple docstring'''


        if key not in self.cache:
          if self.num_keys >= self.capacity:
            # delete first node (oldest) when over capacity
            _a							=	self.list.head.next

            # guaranteed to have a non-None first node when num_keys > 0
            # explain to type checker via assertions
            assert first_node is not None
            assert first_node.key is not None
            assert (
                self.list.remove(a_	) is not None
            )  # node guaranteed to be in list assert node.key is not None

            del self.cache[first_node.key]
            self.num_keys -= 1
          _a							=	DoubleLinkedListNode(a_     ,     a_	)
          self.list.add(self.cache[key]	)
          self.num_keys += 1

        else:
          # bump node to the end of the list, update value
          _a							=	self.list.remove(self.cache[key]	)
          assert node is not None  # node guaranteed to be in list
          _a							=	value
          self.list.add(a_	)






      @classmethod
      def 			lowerCamelCase__    (    cls     ,     a_ = 1_2_8	)	->Callable[[Callable[[T], U]], Callable[..., U]]:







        '''simple docstring'''


        def cache_decorator_inner(a_	) -> Callable[..., U]:
          def cache_decorator_wrapper(*a_	) -> U:
            if func not in cls.decorator_function_to_instance_map:
              _a							=	LRUCache(a_	)

            _a							=	cls.decorator_function_to_instance_map[func].get(args[0]	)
            if result is None:
              _a							=	func(*a_	)
              cls.decorator_function_to_instance_map[func].put(args[0]     ,     a_	)
            return result

          def cache_info() -> LRUCache[T, U]:
            return cls.decorator_function_to_instance_map[func]

          setattr(a_     ,     "cache_info"     ,     a_	)  # noqa: B010

          return cache_decorator_wrapper

        return cache_decorator_inner


if __name__ == "__main__":
     import doctest

     doctest.testmod()


