






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_ :						Optional[Any]      =      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     _SCREAMING_SNAKE_CASE       :





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

		UpperCamelCase     :Union[str, Any]          =	context_length
		UpperCamelCase     :Tuple          =	prediction_length + label_length
		UpperCamelCase     :Tuple          =	label_length

		UpperCamelCase     :List[str]          =	moving_average
		UpperCamelCase     :Any          =	autocorrelation_factor





	def 						_A (		self				:					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 						_A (		self				:					List[Any] ,  __lowerCamelCase				:					Union[str, Any]	):
		UpperCamelCase     :str          =	config.context_length + max(config.lags_sequence	)

		UpperCamelCase     :Union[str, Any]          =	ids_tensor([self.batch_size, 1] ,  config.cardinality[0]	)
		UpperCamelCase     :int          =	floats_tensor([self.batch_size, _past_length, config.num_time_features]	)
		UpperCamelCase     :Dict          =	floats_tensor([self.batch_size, _past_length]	)
		UpperCamelCase     :Optional[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     :Any          =	floats_tensor([self.batch_size, config.prediction_length]	)

		UpperCamelCase     :Union[str, Any]          =	{
		    """past_values""": past_values,
		    """static_categorical_features""": static_categorical_features,
		    """past_time_features""": past_time_features,
		    """past_observed_mask""": past_observed_mask,
		    """future_time_features""": future_time_features,
		    """future_values""": future_values,
		}
		return inputs_dict





	def 						_A (		self				:					List[Any]	):
		UpperCamelCase     :Optional[int]          =	self.get_config()
		UpperCamelCase     :Dict          =	self.prepare_autoformer_inputs_dict(__lowerCamelCase	)
		return config, inputs_dict





	def 						_A (		self				:					Optional[Any]	):
		UpperCamelCase     ,      UpperCamelCase     :Tuple          =	self.prepare_config_and_inputs()
		return config, inputs_dict





	def 						_A (		self				:					Optional[Any] ,  __lowerCamelCase				:					Dict ,  __lowerCamelCase				:					int	):
		UpperCamelCase     :str          =	AutoformerModel(config=__lowerCamelCase	).to(__lowerCamelCase	).eval()
		UpperCamelCase     :Optional[int]          =	model(**__lowerCamelCase	)

		UpperCamelCase     :str          =	outputs.encoder_last_hidden_state
		UpperCamelCase     :List[Any]          =	outputs.last_hidden_state

		with tempfile.TemporaryDirectory() as tmpdirname:
			UpperCamelCase     :List[Any]          =	model.get_encoder()
			encoder.save_pretrained(__lowerCamelCase	)
			UpperCamelCase     :List[Any]          =	AutoformerEncoder.from_pretrained(__lowerCamelCase	).to(__lowerCamelCase	)

		UpperCamelCase     ,      UpperCamelCase     ,      UpperCamelCase     ,      UpperCamelCase     ,      UpperCamelCase     :str          =	model.create_network_inputs(**__lowerCamelCase	)
		UpperCamelCase     ,      UpperCamelCase     :Any          =	model.decomposition_layer(transformer_inputs[:, : config.context_length, ...]	)

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

		UpperCamelCase     :str          =	(
		    torch.mean(transformer_inputs[:, : config.context_length, ...] ,  dim=1	)
		    .unsqueeze(1	)
		    .repeat(1 ,  config.prediction_length ,  1	)
		)
		UpperCamelCase     :Optional[int]          =	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     :Any          =	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     :Optional[int]          =	model.get_decoder()
			decoder.save_pretrained(__lowerCamelCase	)
			UpperCamelCase     :Optional[Any]          =	AutoformerDecoder.from_pretrained(__lowerCamelCase	).to(__lowerCamelCase	)

		UpperCamelCase     :Optional[int]          =	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     _SCREAMING_SNAKE_CASE       (			_a		,   _a		,   unittest.TestCase      ):
	snake_case__							:     List[Any]     	=  (AutoformerModel, AutoformerForPrediction) if is_torch_available() else ()
	snake_case__							:     str     	=  (AutoformerForPrediction,) if is_torch_available() else ()
	snake_case__							:     List[Any]     	=  {"""feature-extraction""": AutoformerModel} if is_torch_available() else {}
	snake_case__							:     Optional[int]     	=  False
	snake_case__							:     List[str]     	=  False
	snake_case__							:     Any     	=  False
	snake_case__							:     Tuple     	=  False
	snake_case__							:     int     	=  False
	snake_case__							:     Optional[int]     	=  False





	def 						_A (		self				:					Any	):
		UpperCamelCase     :Any          =	AutoformerModelTester(self	)
		UpperCamelCase     :Optional[int]          =	ConfigTester(self ,  config_class=__lowerCamelCase ,  has_text_modality=__lowerCamelCase	)





	def 						_A (		self				:					Dict	):
		self.config_tester.run_common_tests()





	def 						_A (		self				:					Tuple	):
		UpperCamelCase     ,      UpperCamelCase     :Any          =	self.model_tester.prepare_config_and_inputs()
		for model_class in self.all_model_classes:
			UpperCamelCase     :Union[str, Any]          =	model_class(__lowerCamelCase	)

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





	def 						_A (		self				:					str	):
		UpperCamelCase     :Optional[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 						_A (		self				:					List[Any]	):
		pass





	def 						_A (		self				:					Dict	):
		UpperCamelCase     :List[Any]          =	inspect.signature(getattr(__lowerCamelCase ,  """forward"""	)	)
		# The main input is the name of the argument after `self`
		UpperCamelCase     :Optional[int]          =	list(model_signature.parameters.keys()	)[1]
		self.assertEqual(AutoformerModel.main_input_name ,  __lowerCamelCase	)





	def 						_A (		self				:					List[Any]	):
		UpperCamelCase     ,      UpperCamelCase     :Optional[int]          =	self.model_tester.prepare_config_and_inputs_for_common()

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

			UpperCamelCase     :List[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(__lowerCamelCase	)] ,  __lowerCamelCase	)





	def 						_A (		self				:					Tuple	):
		UpperCamelCase     ,      UpperCamelCase     :Union[str, Any]          =	self.model_tester.prepare_config_and_inputs_for_common()
		UpperCamelCase     :Dict          =	True

		UpperCamelCase     :Optional[int]          =	getattr(self.model_tester ,  """seq_length""" ,  __lowerCamelCase	)
		UpperCamelCase     :Union[str, Any]          =	getattr(self.model_tester ,  """decoder_seq_length""" ,  __lowerCamelCase	)
		UpperCamelCase     :List[str]          =	getattr(self.model_tester ,  """encoder_seq_length""" ,  __lowerCamelCase	)
		UpperCamelCase     :str          =	getattr(self.model_tester ,  """d_model""" ,  __lowerCamelCase	)
		UpperCamelCase     :str          =	getattr(self.model_tester ,  """num_attention_heads""" ,  __lowerCamelCase	)
		UpperCamelCase     :List[Any]          =	d_model // num_attention_heads

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

			UpperCamelCase     :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
			UpperCamelCase     :Optional[int]          =	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
			UpperCamelCase     :Union[str, 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
		UpperCamelCase     :Dict          =	True
		UpperCamelCase     :Optional[Any]          =	True
		UpperCamelCase     :Tuple          =	model_class(__lowerCamelCase	)
		model.to(__lowerCamelCase	)
		model.eval()
		with torch.no_grad():
			UpperCamelCase     :str          =	model(**self._prepare_for_class(__lowerCamelCase ,  __lowerCamelCase	)	)

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

		UpperCamelCase     :Union[str, 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 						_A (		self				:					str	):
		super().test_retain_grad_hidden_states_attentions()

def    SCREAMING_SNAKE_CASE_							(							__magic_name__					:  Dict="train-batch.pt"					)		->			int:






	"""simple docstring"""



	UpperCamelCase     :Tuple          =	hf_hub_download(repo_id="""hf-internal-testing/tourism-monthly-batch"""	,			filename=__magic_name__	,			repo_type="""dataset"""					)
	UpperCamelCase     :Optional[int]          =	torch.load(__magic_name__	,			map_location=__magic_name__					)
	return batch

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





	def 						_A (		self				:					str	):
		UpperCamelCase     :Optional[Any]          =	AutoformerModel.from_pretrained("""huggingface/autoformer-tourism-monthly"""	).to(__lowerCamelCase	)
		UpperCamelCase     :int          =	prepare_batch()

		with torch.no_grad():
			UpperCamelCase     :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     :int          =	torch.Size(
		    (64, model.config.prediction_length + model.config.label_length, model.config.feature_size)	)
		self.assertEqual(output.shape ,  __lowerCamelCase	)

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





	def 						_A (		self				:					Union[str, Any]	):
		UpperCamelCase     :str          =	AutoformerForPrediction.from_pretrained("""huggingface/autoformer-tourism-monthly"""	).to(__lowerCamelCase	)
		UpperCamelCase     :Any          =	prepare_batch("""val-batch.pt"""	)
		with torch.no_grad():
			UpperCamelCase     :Optional[Any]          =	model(
			    past_values=batch["""past_values"""] ,  past_time_features=batch["""past_time_features"""] ,  past_observed_mask=batch["""past_observed_mask"""] ,  static_categorical_features=batch["""static_categorical_features"""] ,  ).encoder_last_hidden_state
		UpperCamelCase     :str          =	torch.Size((64, model.config.context_length, model.config.d_model)	)
		self.assertEqual(output.shape ,  __lowerCamelCase	)

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





	def 						_A (		self				:					Dict	):
		UpperCamelCase     :List[str]          =	AutoformerForPrediction.from_pretrained("""huggingface/autoformer-tourism-monthly"""	).to(__lowerCamelCase	)
		UpperCamelCase     :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     :Union[str, Any]          =	torch.Size((64, model.config.num_parallel_samples, model.config.prediction_length)	)
		self.assertEqual(outputs.sequences.shape ,  __lowerCamelCase	)

		UpperCamelCase     :Any          =	torch.tensor([3130.6763, 4056.5293, 7053.0786] ,  device=__lowerCamelCase	)
		UpperCamelCase     :Union[str, Any]          =	outputs.sequences.mean(dim=1	)
		self.assertTrue(torch.allclose(mean_prediction[0, -3:] ,  __lowerCamelCase ,  rtol=1E-1	)	)





# Copyright 2023 The HuggingFace Inc. team. All rights reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
from ..models.auto import AutoModelForSeqaSeqLM, AutoTokenizer
from .base import PipelineTool




class 					snake_case__						(       lowerCAmelCase_     ):






     """simple docstring"""


     _SCREAMING_SNAKE_CASE						=	"""philschmid/bart-large-cnn-samsum"""
     _SCREAMING_SNAKE_CASE						=	(
         """This is a tool that summarizes an English text. It takes an input `text` containing the text to summarize, """
         """and returns a summary of the text."""
     )
     _SCREAMING_SNAKE_CASE						=	"""summarizer"""
     _SCREAMING_SNAKE_CASE						=	AutoTokenizer
     _SCREAMING_SNAKE_CASE						=	AutoModelForSeqaSeqLM

     _SCREAMING_SNAKE_CASE						=	["""text"""]
     _SCREAMING_SNAKE_CASE						=	["""text"""]






     def    lowercase_   (       self  :  Optional[Any], _snake_case  :  str					)   ->Any:
            return self.pre_processor(_snake_case, return_tensors='pt', truncation=_snake_case					)






     def    lowercase_   (       self  :  int, _snake_case  :  List[Any]					)   ->Any:
            return self.model.generate(**_snake_case					)[0]






     def    lowercase_   (       self  :  int, _snake_case  :  int					)   ->str:
            return self.pre_processor.decode(_snake_case, skip_special_tokens=_snake_case, clean_up_tokenization_spaces=_snake_case					)






