


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


_lowerCAmelCase     :							List[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 		snake_case				:

				"""simple docstring"""




				def __init__(	self       ,    lowerCamelCase       ,    lowerCamelCase=16       ,    lowerCamelCase=13       ,    lowerCamelCase=7       ,    lowerCamelCase=14       ,    lowerCamelCase=10       ,    lowerCamelCase=19       ,    lowerCamelCase=5       ,    lowerCamelCase=4       ,    lowerCamelCase=True       ,    lowerCamelCase=16       ,    lowerCamelCase=2       ,    lowerCamelCase=4       ,    lowerCamelCase=4       ,    lowerCamelCase="gelu"       ,    lowerCamelCase=0.1       ,    lowerCamelCase=0.1       ,    lowerCamelCase=[1, 2, 3, 4, 5]       ,    lowerCamelCase=25       ,    lowerCamelCase=5       ,    )    ->      List[str]:







					"""simple docstring"""






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

					snake_case__			:			Any             =		context_length
					snake_case__			:			List[Any]             =		prediction_length + label_length
					snake_case__			:			Optional[int]             =		label_length

					snake_case__			:			Union[str, Any]             =		moving_average
					snake_case__			:			Any             =		autocorrelation_factor





				def    lowercase__							(	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    lowercase__							(	self       ,    lowerCamelCase )    ->      List[Any]:







					"""simple docstring"""






					snake_case__			:			int             =		config.context_length + max(config.lags_sequence )

					snake_case__			:			Dict             =		ids_tensor([self.batch_size, 1]       ,    config.cardinality[0] )
					snake_case__			:			Tuple             =		floats_tensor([self.batch_size, _past_length, config.num_time_features] )
					snake_case__			:			Any             =		floats_tensor([self.batch_size, _past_length] )
					snake_case__			:			int             =		floats_tensor([self.batch_size, _past_length] ) > 0.5

					# decoder inputs
					snake_case__			:			str             =		floats_tensor([self.batch_size, config.prediction_length, config.num_time_features] )
					snake_case__			:			int             =		floats_tensor([self.batch_size, config.prediction_length] )

					snake_case__			:			List[str]             =		{
					    '''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 )    ->      Any:







					"""simple docstring"""






					snake_case__			:			Any             =		self.get_config()
					snake_case__			:			Tuple             =		self.prepare_autoformer_inputs_dict(lowerCamelCase )
					return config, inputs_dict





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







					"""simple docstring"""






					snake_case__      ,snake_case__			:			str             =		self.prepare_config_and_inputs()
					return config, inputs_dict





				def    lowercase__							(	self       ,    lowerCamelCase       ,    lowerCamelCase )    ->      Union[str, Any]:







					"""simple docstring"""






					snake_case__			:			str             =		AutoformerModel(config=lowerCamelCase ).to(lowerCamelCase ).eval()
					snake_case__			:			Tuple             =		model(**lowerCamelCase )

					snake_case__			:			Dict             =		outputs.encoder_last_hidden_state
					snake_case__			:			Optional[Any]             =		outputs.last_hidden_state

					with tempfile.TemporaryDirectory() as tmpdirname:
						snake_case__			:			int             =		model.get_encoder()
						encoder.save_pretrained(lowerCamelCase )
						snake_case__			:			Union[str, Any]             =		AutoformerEncoder.from_pretrained(lowerCamelCase ).to(lowerCamelCase )

					snake_case__      ,snake_case__      ,snake_case__      ,snake_case__      ,snake_case__			:			Dict             =		model.create_network_inputs(**lowerCamelCase )
					snake_case__      ,snake_case__			:			str             =		model.decomposition_layer(transformer_inputs[:, : config.context_length, ...] )

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

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

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

					snake_case__			:			Optional[Any]             =		decoder(
					    trend=lowerCamelCase       ,    inputs_embeds=lowerCamelCase       ,    encoder_hidden_states=lowerCamelCase       ,    )[0]

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



@require_torch
class 		snake_case				(    __lowerCamelCase				,    __lowerCamelCase				,    unittest.TestCase						):

				"""simple docstring"""




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





				def    lowercase__							(	self )    ->      Union[str, Any]:







					"""simple docstring"""






					snake_case__			:			Any             =		AutoformerModelTester(self )
					snake_case__			:			Dict             =		ConfigTester(self       ,    config_class=lowerCamelCase       ,    has_text_modality=lowerCamelCase )





				def    lowercase__							(	self )    ->      Dict:







					"""simple docstring"""






					self.config_tester.run_common_tests()





				def    lowercase__							(	self )    ->      Dict:







					"""simple docstring"""






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

						with tempfile.TemporaryDirectory() as tmpdirname:
							model.save_pretrained(lowerCamelCase )
							snake_case__      ,snake_case__			:			List[Any]             =		model_class.from_pretrained(lowerCamelCase       ,    output_loading_info=lowerCamelCase )
						self.assertEqual(info['''missing_keys''']       ,    [] )





				def    lowercase__							(	self )    ->      str:







					"""simple docstring"""






					snake_case__			:			Union[str, Any]             =		self.model_tester.prepare_config_and_inputs_for_common()
					self.model_tester.check_encoder_decoder_model_standalone(*lowerCamelCase )





				@unittest.skip(reason='''Model has no tokens embeddings''' )
				def    lowercase__							(	self )    ->      Tuple:







					"""simple docstring"""






					pass





				def    lowercase__							(	self )    ->      Union[str, Any]:







					"""simple docstring"""






					snake_case__			:			Optional[Any]             =		inspect.signature(getattr(lowerCamelCase       ,    '''forward''' ) )
					# The main input is the name of the argument after `self`
					snake_case__			:			Dict             =		list(model_signature.parameters.keys() )[1]
					self.assertEqual(AutoformerModel.main_input_name       ,    lowerCamelCase )





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







					"""simple docstring"""






					snake_case__      ,snake_case__			:			Dict             =		self.model_tester.prepare_config_and_inputs_for_common()

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

						snake_case__			:			int             =		[
						    '''past_values''',
						    '''past_time_features''',
						    '''past_observed_mask''',
						    '''static_categorical_features''',
						    '''static_real_features''',
						    '''future_values''',
						    '''future_time_features''',
						]

						if model.__class__.__name__ in ["AutoformerForPrediction"]:
							expected_arg_names.append('''future_observed_mask''' )

						expected_arg_names.extend(
						    [
						        '''decoder_attention_mask''',
						        '''head_mask''',
						        '''decoder_head_mask''',
						        '''cross_attn_head_mask''',
						        '''encoder_outputs''',
						        '''past_key_values''',
						        '''output_hidden_states''',
						        '''output_attentions''',
						        '''use_cache''',
						        '''return_dict''',
						    ] )

						self.assertListEqual(arg_names[: len(lowerCamelCase )]       ,    lowerCamelCase )





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







					"""simple docstring"""






					snake_case__      ,snake_case__			:			Optional[int]             =		self.model_tester.prepare_config_and_inputs_for_common()
					snake_case__			:			Any             =		True

					snake_case__			:			Tuple             =		getattr(self.model_tester       ,    '''seq_length'''       ,    lowerCamelCase )
					snake_case__			:			str             =		getattr(self.model_tester       ,    '''decoder_seq_length'''       ,    lowerCamelCase )
					snake_case__			:			int             =		getattr(self.model_tester       ,    '''encoder_seq_length'''       ,    lowerCamelCase )
					snake_case__			:			Any             =		getattr(self.model_tester       ,    '''d_model'''       ,    lowerCamelCase )
					snake_case__			:			List[Any]             =		getattr(self.model_tester       ,    '''num_attention_heads'''       ,    lowerCamelCase )
					snake_case__			:			Tuple             =		d_model // num_attention_heads

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

						# check that output_attentions also work using config
						del inputs_dict["output_attentions"]
						snake_case__			:			str             =		True
						snake_case__			:			List[str]             =		model_class(lowerCamelCase )
						model.to(lowerCamelCase )
						model.eval()
						with torch.no_grad():
							snake_case__			:			int             =		model(**self._prepare_for_class(lowerCamelCase       ,    lowerCamelCase ) )
						snake_case__			:			Optional[Any]             =		outputs.encoder_attentions
						self.assertEqual(len(lowerCamelCase )       ,    self.model_tester.num_hidden_layers )

						self.assertListEqual(
						    list(attentions[0].shape[-3:] )       ,    [self.model_tester.num_attention_heads, encoder_seq_length, dim]       ,    )
						snake_case__			:			int             =		len(lowerCamelCase )

						snake_case__			:			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(lowerCamelCase       ,    lowerCamelCase )

						# decoder attentions
						snake_case__			:			List[Any]             =		outputs.decoder_attentions
						self.assertIsInstance(lowerCamelCase       ,    (list, tuple) )
						self.assertEqual(len(lowerCamelCase )       ,    self.model_tester.num_hidden_layers )
						self.assertListEqual(
						    list(decoder_attentions[0].shape[-3:] )       ,    [self.model_tester.num_attention_heads, decoder_seq_length, dim]       ,    )

						# cross attentions
						snake_case__			:			Optional[Any]             =		outputs.cross_attentions
						self.assertIsInstance(lowerCamelCase       ,    (list, tuple) )
						self.assertEqual(len(lowerCamelCase )       ,    self.model_tester.num_hidden_layers )
						self.assertListEqual(
						    list(cross_attentions[0].shape[-3:] )       ,    [self.model_tester.num_attention_heads, decoder_seq_length, dim]       ,    )

					# Check attention is always last and order is fine
					snake_case__			:			int             =		True
					snake_case__			:			Tuple             =		True
					snake_case__			:			Dict             =		model_class(lowerCamelCase )
					model.to(lowerCamelCase )
					model.eval()
					with torch.no_grad():
						snake_case__			:			Optional[Any]             =		model(**self._prepare_for_class(lowerCamelCase       ,    lowerCamelCase ) )

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

					snake_case__			:			Any             =		outputs.encoder_attentions if config.is_encoder_decoder else outputs.attentions

					self.assertEqual(len(lowerCamelCase )       ,    self.model_tester.num_hidden_layers )
					self.assertListEqual(
					    list(self_attentions[0].shape[-3:] )       ,    [self.model_tester.num_attention_heads, encoder_seq_length, dim]       ,    )





				@is_flaky()
				def    lowercase__							(	self )    ->      Optional[Any]:







					"""simple docstring"""






					super().test_retain_grad_hidden_states_attentions()






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





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

				"""simple docstring"""




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







					"""simple docstring"""






					snake_case__			:			Dict             =		AutoformerModel.from_pretrained('''huggingface/autoformer-tourism-monthly''' ).to(lowerCamelCase )
					snake_case__			:			Any             =		prepare_batch()

					with torch.no_grad():
						snake_case__			:			List[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''']       ,    future_values=batch['''future_values''']       ,    future_time_features=batch['''future_time_features''']       ,    )[0]

					snake_case__			:			List[Any]             =		torch.Size(
					    (64, model.config.prediction_length + model.config.label_length, model.config.feature_size) )
					self.assertEqual(output.shape       ,    lowerCamelCase )

					snake_case__			:			Optional[int]             =		torch.tensor(
					    [[0.3_593, -1.3_398, 0.6_330], [0.2_279, 1.5_396, -0.1_792], [0.0_450, 1.3_225, -0.2_335]]       ,    device=lowerCamelCase )
					self.assertTrue(torch.allclose(output[0, :3, :3]       ,    lowerCamelCase       ,    atol=lowerCamelCase ) )





				def    lowercase__							(	self )    ->      Any:







					"""simple docstring"""






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

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





				def    lowercase__							(	self )    ->      Tuple:







					"""simple docstring"""






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

					snake_case__			:			Union[str, Any]             =		torch.tensor([3_130.6_763, 4_056.5_293, 7_053.0_786]       ,    device=lowerCamelCase )
					snake_case__			:			Dict             =		outputs.sequences.mean(dim=1 )
					self.assertTrue(torch.allclose(mean_prediction[0, -3:]       ,    lowerCamelCase       ,    rtol=1E-1 ) )










'''simple docstring'''





from typing import TYPE_CHECKING

from ...utils import (
    OptionalDependencyNotAvailable,
    _LazyModule,
    is_tf_available,
    is_torch_available,
    is_vision_available,
)


lowercase_									=    {
    "configuration_blip": [
        "BLIP_PRETRAINED_CONFIG_ARCHIVE_MAP",
        "BlipConfig",
        "BlipTextConfig",
        "BlipVisionConfig",
    ],
    "processing_blip": ["BlipProcessor"],
}

try:
  if not is_vision_available():
    raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
  pass
else:
  lowercase_									=    ["BlipImageProcessor"]


try:
  if not is_torch_available():
    raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
  pass
else:
  lowercase_									=    [
      "BLIP_PRETRAINED_MODEL_ARCHIVE_LIST",
      "BlipModel",
      "BlipPreTrainedModel",
      "BlipForConditionalGeneration",
      "BlipForQuestionAnswering",
      "BlipVisionModel",
      "BlipTextModel",
      "BlipForImageTextRetrieval",
  ]

try:
  if not is_tf_available():
    raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
  pass
else:
  lowercase_									=    [
      "TF_BLIP_PRETRAINED_MODEL_ARCHIVE_LIST",
      "TFBlipModel",
      "TFBlipPreTrainedModel",
      "TFBlipForConditionalGeneration",
      "TFBlipForQuestionAnswering",
      "TFBlipVisionModel",
      "TFBlipTextModel",
      "TFBlipForImageTextRetrieval",
  ]

if TYPE_CHECKING:
  from .configuration_blip import BLIP_PRETRAINED_CONFIG_ARCHIVE_MAP, BlipConfig, BlipTextConfig, BlipVisionConfig
  from .processing_blip import BlipProcessor

  try:
    if not is_vision_available():
      raise OptionalDependencyNotAvailable()
  except OptionalDependencyNotAvailable:
    pass
  else:
    from .image_processing_blip import BlipImageProcessor

  try:
    if not is_torch_available():
      raise OptionalDependencyNotAvailable()
  except OptionalDependencyNotAvailable:
    pass
  else:
    from .modeling_blip import (
        BLIP_PRETRAINED_MODEL_ARCHIVE_LIST,
        BlipForConditionalGeneration,
        BlipForImageTextRetrieval,
        BlipForQuestionAnswering,
        BlipModel,
        BlipPreTrainedModel,
        BlipTextModel,
        BlipVisionModel,
    )

  try:
    if not is_tf_available():
      raise OptionalDependencyNotAvailable()
  except OptionalDependencyNotAvailable:
    pass
  else:
    from .modeling_tf_blip import (
        TF_BLIP_PRETRAINED_MODEL_ARCHIVE_LIST,
        TFBlipForConditionalGeneration,
        TFBlipForImageTextRetrieval,
        TFBlipForQuestionAnswering,
        TFBlipModel,
        TFBlipPreTrainedModel,
        TFBlipTextModel,
        TFBlipVisionModel,
    )

else:
  import sys

  lowercase_									=    _LazyModule(__name__, globals()["__file__"], _import_structure, module_spec=__spec__)



