


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


snake_case							:	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 	snake_case_       :
					def __init__(				self       :Tuple      ,__snake_case       :List[Any]      ,__snake_case       :Optional[int]=16      ,__snake_case       :Union[str, Any]=13      ,__snake_case       :List[Any]=7      ,__snake_case       :Optional[Any]=14      ,__snake_case       :str=10      ,__snake_case       :Dict=19      ,__snake_case       :Dict=5      ,__snake_case       :Optional[int]=4      ,__snake_case       :List[str]=True      ,__snake_case       :List[str]=16      ,__snake_case       :Any=2      ,__snake_case       :Optional[int]=4      ,__snake_case       :Optional[int]=4      ,__snake_case       :Union[str, Any]="gelu"      ,__snake_case       :Dict=0.1      ,__snake_case       :int=0.1      ,__snake_case       :Optional[Any]=[1, 2, 3, 4, 5]      ,__snake_case       :List[str]=25      ,__snake_case       :str=5      ,) ->   str:
												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       :Union[str, Any]				) ->   Union[str, Any]:
												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       :Optional[Any]      ,__snake_case       :Tuple				) ->   Dict:
												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       :List[str]				) ->   Tuple:
												a__					=					self.get_config()
												a__					=					self.prepare_autoformer_inputs_dict(__snake_case				)
												return config, inputs_dict
					def 			lowerCamelCase__(				self       :Dict				) ->   Union[str, Any]:
												a__				, a__					=					self.prepare_config_and_inputs()
												return config, inputs_dict



					def 			lowerCamelCase__(				self       :str      ,__snake_case       :str      ,__snake_case       :Tuple				) ->   List[str]:
												a__					=					AutoformerModel(config=__snake_case				).to(__snake_case				).eval()
												a__					=					model(**__snake_case				)

												a__					=					outputs.encoder_last_hidden_state
												a__					=					outputs.last_hidden_state

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

												a__				, a__				, a__				, a__				, a__					=					model.create_network_inputs(**__snake_case				)
												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=__snake_case				)[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(__snake_case				)
																			a__					=					AutoformerDecoder.from_pretrained(__snake_case				).to(__snake_case				)

												a__					=					decoder(
												    trend=__snake_case      ,inputs_embeds=__snake_case      ,encoder_hidden_states=__snake_case      ,)[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				):
					UpperCAmelCase__ :   Tuple   				=      (AutoformerModel, AutoformerForPrediction) if is_torch_available() else ()
					UpperCAmelCase__ :   Optional[Any]   				=      (AutoformerForPrediction,) if is_torch_available() else ()
					UpperCAmelCase__ :   Dict   				=      {'''feature-extraction''': AutoformerModel} if is_torch_available() else {}
					UpperCAmelCase__ :   Optional[Any]   				=      False
					UpperCAmelCase__ :   Tuple   				=      False
					UpperCAmelCase__ :   str   				=      False
					UpperCAmelCase__ :   List[str]   				=      False
					UpperCAmelCase__ :   Optional[Any]   				=      False
					UpperCAmelCase__ :   List[Any]   				=      False
					def 			lowerCamelCase__(				self       :Dict				) ->   List[str]:
												a__					=					AutoformerModelTester(self				)
												a__					=					ConfigTester(self      ,config_class=__snake_case      ,has_text_modality=__snake_case				)
					def 			lowerCamelCase__(				self       :str				) ->   Any:
												self.config_tester.run_common_tests()
					def 			lowerCamelCase__(				self       :Optional[Any]				) ->   int:
												a__				, a__					=					self.model_tester.prepare_config_and_inputs()
												for model_class in self.all_model_classes:
																			a__					=					model_class(__snake_case				)

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

												for model_class in self.all_model_classes:
																			a__					=					model_class(__snake_case				)
																			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(__snake_case				)]      ,__snake_case				)
					def 			lowerCamelCase__(				self       :Union[str, Any]				) ->   Dict:
												a__				, a__					=					self.model_tester.prepare_config_and_inputs_for_common()
												a__					=					True

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

												for model_class in self.all_model_classes:
																			a__					=					True
																			a__					=					False
																			a__					=					True
																			a__					=					model_class(__snake_case				)
																			model.to(__snake_case				)
																			model.eval()
																			with torch.no_grad():
																										a__					=					model(**self._prepare_for_class(__snake_case      ,__snake_case				)				)
																			a__					=					outputs.encoder_attentions if config.is_encoder_decoder else outputs.attentions
																			self.assertEqual(len(__snake_case				)      ,self.model_tester.num_hidden_layers				)

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

																			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(__snake_case      ,__snake_case				)

																			# decoder attentions
																			a__					=					outputs.decoder_attentions
																			self.assertIsInstance(__snake_case      ,(list, tuple)				)
																			self.assertEqual(len(__snake_case				)      ,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(__snake_case      ,(list, tuple)				)
																			self.assertEqual(len(__snake_case				)      ,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(__snake_case				)
												model.to(__snake_case				)
												model.eval()
												with torch.no_grad():
																			a__					=					model(**self._prepare_for_class(__snake_case      ,__snake_case				)				)

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

												a__					=					outputs.encoder_attentions if config.is_encoder_decoder else outputs.attentions

												self.assertEqual(len(__snake_case				)      ,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       :List[str]				) ->   Any:
												super().test_retain_grad_hidden_states_attentions()






def 							__lowercase       (			__lowerCAmelCase							:    Dict="train-batch.pt"       ):
							a__					=					hf_hub_download(repo_id='hf-internal-testing/tourism-monthly-batch' ,   filename=__lowerCAmelCase ,   repo_type='dataset'       )
							a__					=					torch.load(__lowerCAmelCase ,   map_location=__lowerCAmelCase       )
							return batch







@require_torch
@slow
class 	snake_case_       (unittest.TestCase				):
					def 			lowerCamelCase__(				self       :Any				) ->   str:
												a__					=					AutoformerModel.from_pretrained('huggingface/autoformer-tourism-monthly'				).to(__snake_case				)
												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(
												    (64, model.config.prediction_length + model.config.label_length, model.config.feature_size)				)
												self.assertEqual(output.shape      ,__snake_case				)

												a__					=					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=__snake_case				)
												self.assertTrue(torch.allclose(output[0, :3, :3]      ,__snake_case      ,atol=__snake_case				)				)
					def 			lowerCamelCase__(				self       :List[Any]				) ->   Optional[Any]:
												a__					=					AutoformerForPrediction.from_pretrained('huggingface/autoformer-tourism-monthly'				).to(__snake_case				)
												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((64, model.config.context_length, model.config.d_model)				)
												self.assertEqual(output.shape      ,__snake_case				)

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



					def 			lowerCamelCase__(				self       :List[Any]				) ->   int:
												a__					=					AutoformerForPrediction.from_pretrained('huggingface/autoformer-tourism-monthly'				).to(__snake_case				)
												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((64, model.config.num_parallel_samples, model.config.prediction_length)				)
												self.assertEqual(outputs.sequences.shape      ,__snake_case				)

												a__					=					torch.tensor([31_30.67_63, 40_56.52_93, 70_53.07_86]      ,device=__snake_case				)
												a__					=					outputs.sequences.mean(dim=1				)
												self.assertTrue(torch.allclose(mean_prediction[0, -3:]      ,__snake_case      ,rtol=1E-1				)				)








from typing import Dict, List, Optional, Union

import numpy as np

from ...image_processing_utils import BaseImageProcessor, BatchFeature, get_size_dict
from ...image_transforms import rescale, resize, to_channel_dimension_format
from ...image_utils import (
    ChannelDimension,
    ImageInput,
    PILImageResampling,
    make_list_of_images,
    to_numpy_array,
    valid_images,
)
from ...utils import TensorType, is_vision_available, logging


if is_vision_available():
						import PIL


snake_case							:	str					      =				logging.get_logger(__name__)






def 							__lowercase       (			__lowerCAmelCase							:    Union[str, Any] ,   __lowerCAmelCase							:    Optional[Any]       ):
							a__					=					b.T
							a__					=					np.sum(np.square(__lowerCAmelCase       ) ,   axis=1       )
							a__					=					np.sum(np.square(__lowerCAmelCase       ) ,   axis=0       )
							a__					=					np.matmul(__lowerCAmelCase ,   __lowerCAmelCase       )
							a__					=					aa[:, None] - 2 * ab + ba[None, :]
							return d






def 							__lowercase       (			__lowerCAmelCase							:    List[Any] ,   __lowerCAmelCase							:    str       ):
							a__					=					x.reshape(-1 ,   3       )
							a__					=					squared_euclidean_distance(__lowerCAmelCase ,   __lowerCAmelCase       )
							return np.argmin(__lowerCAmelCase ,   axis=1       )







class 	snake_case_       (lowerCamelCase_				):
					UpperCAmelCase__ :   Optional[Any]   				=      ['''pixel_values''']
					def __init__(				self       :Dict      ,__snake_case       :Optional[Union[List[List[int]], np.ndarray]] = None      ,__snake_case       :bool = True      ,__snake_case       :Dict[str, int] = None      ,__snake_case       :PILImageResampling = PILImageResampling.BILINEAR      ,__snake_case       :bool = True      ,__snake_case       :bool = True      ,**__snake_case       :Optional[int]      ,) ->   None:
												super().__init__(**__snake_case				)
												a__					=					size if size is not None else {'height': 2_56, 'width': 2_56}
												a__					=					get_size_dict(__snake_case				)
												a__					=					np.array(__snake_case				) if clusters is not None else None
												a__					=					do_resize
												a__					=					size
												a__					=					resample
												a__					=					do_normalize
												a__					=					do_color_quantize
					def 			lowerCamelCase__(				self       :Union[str, Any]      ,__snake_case       :np.ndarray      ,__snake_case       :Dict[str, int]      ,__snake_case       :PILImageResampling = PILImageResampling.BILINEAR      ,__snake_case       :Optional[Union[str, ChannelDimension]] = None      ,**__snake_case       :Any      ,) ->   np.ndarray:
												a__					=					get_size_dict(__snake_case				)
												if "height" not in size or "width" not in size:
																			raise ValueError(F'Size dictionary must contain both height and width keys. Got {size.keys()}'				)
												return resize(
												    __snake_case      ,size=(size['height'], size['width'])      ,resample=__snake_case      ,data_format=__snake_case      ,**__snake_case				)
					def 			lowerCamelCase__(				self       :List[Any]      ,__snake_case       :np.ndarray      ,__snake_case       :Optional[Union[str, ChannelDimension]] = None      ,) ->   np.ndarray:
												a__					=					rescale(image=__snake_case      ,scale=1 / 1_27.5      ,data_format=__snake_case				)
												a__					=					image - 1
												return image



					def 			lowerCamelCase__(				self       :Optional[Any]      ,__snake_case       :ImageInput      ,__snake_case       :bool = None      ,__snake_case       :Dict[str, int] = None      ,__snake_case       :PILImageResampling = None      ,__snake_case       :bool = None      ,__snake_case       :Optional[bool] = None      ,__snake_case       :Optional[Union[List[List[int]], np.ndarray]] = None      ,__snake_case       :Optional[Union[str, TensorType]] = None      ,__snake_case       :Optional[Union[str, ChannelDimension]] = ChannelDimension.FIRST      ,**__snake_case       :List[str]      ,) ->   PIL.Image.Image:
												a__					=					do_resize if do_resize is not None else self.do_resize
												a__					=					size if size is not None else self.size
												a__					=					get_size_dict(__snake_case				)
												a__					=					resample if resample is not None else self.resample
												a__					=					do_normalize if do_normalize is not None else self.do_normalize
												a__					=					do_color_quantize if do_color_quantize is not None else self.do_color_quantize
												a__					=					clusters if clusters is not None else self.clusters
												a__					=					np.array(__snake_case				)

												a__					=					make_list_of_images(__snake_case				)

												if not valid_images(__snake_case				):
																			raise ValueError(
																			    'Invalid image type. Must be of type PIL.Image.Image, numpy.ndarray, '
																			    'torch.Tensor, tf.Tensor or jax.ndarray.'				)

												if do_resize and size is None or resample is None:
																			raise ValueError('Size and resample must be specified if do_resize is True.'				)

												if do_color_quantize and clusters is None:
																			raise ValueError('Clusters must be specified if do_color_quantize is True.'				)

												# All transformations expect numpy arrays.
												a__					=					[to_numpy_array(__snake_case				) for image in images]

												if do_resize:
																			a__					=					[self.resize(image=__snake_case      ,size=__snake_case      ,resample=__snake_case				) for image in images]

												if do_normalize:
																			a__					=					[self.normalize(image=__snake_case				) for image in images]

												if do_color_quantize:
																			a__					=					[to_channel_dimension_format(__snake_case      ,ChannelDimension.LAST				) for image in images]
																			# color quantize from (batch_size, height, width, 3) to (batch_size, height, width)
																			a__					=					np.array(__snake_case				)
																			a__					=					color_quantize(__snake_case      ,__snake_case				).reshape(images.shape[:-1]				)

																			# flatten to (batch_size, height*width)
																			a__					=					images.shape[0]
																			a__					=					images.reshape(__snake_case      ,-1				)

																			# We need to convert back to a list of images to keep consistent behaviour across processors.
																			a__					=					list(__snake_case				)
												else:
																			a__					=					[to_channel_dimension_format(__snake_case      ,__snake_case				) for image in images]

												a__					=					{'input_ids': images}
												return BatchFeature(data=__snake_case      ,tensor_type=__snake_case				)




