






import os
import sys
import warnings
from dataclasses import dataclass, field
from io import BytesIO
from typing import TYPE_CHECKING, Any, ClassVar, Dict, List, Optional, Union

import numpy as np
import pyarrow as pa

from .. import config
from ..download.streaming_download_manager import xopen
from ..table import array_cast
from ..utils.file_utils import is_local_path
from ..utils.py_utils import first_non_null_value, no_op_if_value_is_null, string_to_dict


if TYPE_CHECKING:
	import PIL.Image

	from .features import FeatureType


lowerCamelCase : Optional[List[str]]						   =       None
lowerCamelCase : List[str]						   =       '''<''' if sys.byteorder == '''little''' else '''>'''
# Origin: https://github.com/python-pillow/Pillow/blob/698951e19e19972aeed56df686868f1329981c12/src/PIL/Image.py#L3126 minus "|i1" which values are not preserved correctly when saving and loading an image
lowerCamelCase : List[str]						   =       [
    np.dtype('''|b1'''),
    np.dtype('''|u1'''),
    np.dtype('''<u2'''),
    np.dtype('''>u2'''),
    np.dtype('''<i2'''),
    np.dtype('''>i2'''),
    np.dtype('''<u4'''),
    np.dtype('''>u4'''),
    np.dtype('''<i4'''),
    np.dtype('''>i4'''),
    np.dtype('''<f4'''),
    np.dtype('''>f4'''),
    np.dtype('''<f8'''),
    np.dtype('''>f8'''),
]

@dataclass
class 		lowerCAmelCase     :

					'''simple docstring'''

					_A						:			bool									=      True
					_A						:			Optional[str]									=      None
					# Automatically constructed
					_A						:			ClassVar[str]									=      "PIL.Image.Image"
					_A						:			ClassVar[Any]									=      pa.struct({'''bytes''': pa.binary(), '''path''': pa.string()}			)
					_A						:			str									=      field(default='''Image'''      ,       init=__a      ,       repr=__a			)



					def __call__(  self				:	Tuple							)			->  Tuple:


												"""simple docstring"""

												return self.pa_type



					def       lowerCAmelCase				(  self				:	Optional[Any]		,     __a				:	Union[str, bytes, dict, np.ndarray, "PIL.Image.Image"]							)			->  dict:


												"""simple docstring"""

												if config.PIL_AVAILABLE:
																			import PIL.Image
												else:
																			raise ImportError("""To support encoding images, please install 'Pillow'."""							)

												if isinstance(__a		,     __a							):
																			__lowercase      :							str				=     np.array(__a							)

												if isinstance(__a		,     __a							):
																			return {"path": value, "bytes": None}
												elif isinstance(__a		,     __a							):
																			return {"path": None, "bytes": value}
												elif isinstance(__a		,     np.ndarray							):
																			# convert the image array to PNG/TIFF bytes
																			return encode_np_array(__a							)
												elif isinstance(__a		,     PIL.Image.Image							):
																			# convert the PIL image to bytes (default format is PNG/TIFF)
																			return encode_pil_image(__a							)
												elif value.get("""path"""							) is not None and os.path.isfile(value["""path"""]							):
																			# we set "bytes": None to not duplicate the data if they're already available locally
																			return {"bytes": None, "path": value.get("""path"""							)}
												elif value.get("""bytes"""							) is not None or value.get("""path"""							) is not None:
																			# store the image bytes, and path is used to infer the image format using the file extension
																			return {"bytes": value.get("""bytes"""							), "path": value.get("""path"""							)}
												else:
																			raise ValueError(
																			    F"An image sample should have one of 'path' or 'bytes' but they are missing or None in {value}."							)



					def       lowerCAmelCase				(  self				:	List[str]		,     __a				:	dict		,     __a				:	Union[str, Any]=None							)			->  "PIL.Image.Image":


												"""simple docstring"""

												if not self.decode:
																			raise RuntimeError("""Decoding is disabled for this feature. Please use Image(decode=True) instead."""							)

												if config.PIL_AVAILABLE:
																			import PIL.Image
												else:
																			raise ImportError("""To support decoding images, please install 'Pillow'."""							)

												if token_per_repo_id is None:
																			__lowercase      :							Union[str, Any]				=     {}

												__lowercase      :							List[Any]				=     value["""path"""], value["""bytes"""]
												if bytes_ is None:
																			if path is None:
																										raise ValueError(F"An image should have one of 'path' or 'bytes' but both are None in {value}."							)
																			else:
																										if is_local_path(__a							):
																																	__lowercase      :							str				=     PIL.Image.open(__a							)
																										else:
																																	__lowercase      :							Dict				=     path.split("""::"""							)[-1]
																																	try:
																																								__lowercase      :							List[str]				=     string_to_dict(__a		,     config.HUB_DATASETS_URL							)["""repo_id"""]
																																								__lowercase      :							int				=     token_per_repo_id.get(__a							)
																																	except ValueError:
																																								__lowercase      :							List[Any]				=     None
																																	with xopen(__a		,     """rb"""		,     use_auth_token=__a							) as f:
																																								__lowercase      :							Dict				=     BytesIO(f.read()							)
																																	__lowercase      :							int				=     PIL.Image.open(bytes_							)
												else:
																			__lowercase      :							Optional[Any]				=     PIL.Image.open(BytesIO(bytes_							)							)
												image.load()  # to avoid "Too many open files" errors
												return image



					def       lowerCAmelCase				(  self				:	Union[str, Any]							)			->  Union["FeatureType", Dict[str, "FeatureType"]]:


												"""simple docstring"""

												from .features import Value

												return (
												    self
												    if self.decode
												    else {
												        "bytes": Value("""binary"""							),
												        "path": Value("""string"""							),
												    }
												)



					def       lowerCAmelCase				(  self				:	List[str]		,     __a				:	Union[pa.StringArray, pa.StructArray, pa.ListArray]							)			->  pa.StructArray:


												"""simple docstring"""

												if pa.types.is_string(storage.type							):
																			__lowercase      :							Optional[int]				=     pa.array([None] * len(__a							)		,     type=pa.binary()							)
																			__lowercase      :							Optional[Any]				=     pa.StructArray.from_arrays([bytes_array, storage]		,     ["""bytes""", """path"""]		,     mask=storage.is_null()							)
												elif pa.types.is_binary(storage.type							):
																			__lowercase      :							Dict				=     pa.array([None] * len(__a							)		,     type=pa.string()							)
																			__lowercase      :							Optional[int]				=     pa.StructArray.from_arrays([storage, path_array]		,     ["""bytes""", """path"""]		,     mask=storage.is_null()							)
												elif pa.types.is_struct(storage.type							):
																			if storage.type.get_field_index("""bytes"""							) >= 0:
																										__lowercase      :							List[Any]				=     storage.field("""bytes"""							)
																			else:
																										__lowercase      :							Optional[int]				=     pa.array([None] * len(__a							)		,     type=pa.binary()							)
																			if storage.type.get_field_index("""path"""							) >= 0:
																										__lowercase      :							Any				=     storage.field("""path"""							)
																			else:
																										__lowercase      :							Optional[Any]				=     pa.array([None] * len(__a							)		,     type=pa.string()							)
																			__lowercase      :							Optional[int]				=     pa.StructArray.from_arrays([bytes_array, path_array]		,     ["""bytes""", """path"""]		,     mask=storage.is_null()							)
												elif pa.types.is_list(storage.type							):
																			__lowercase      :							Dict				=     pa.array(
																			    [encode_np_array(np.array(__a							)							)["""bytes"""] if arr is not None else None for arr in storage.to_pylist()]		,     type=pa.binary()		,     )
																			__lowercase      :							Dict				=     pa.array([None] * len(__a							)		,     type=pa.string()							)
																			__lowercase      :							Optional[Any]				=     pa.StructArray.from_arrays(
																			    [bytes_array, path_array]		,     ["""bytes""", """path"""]		,     mask=bytes_array.is_null()							)
												return array_cast(__a		,     self.pa_type							)



					def       lowerCAmelCase				(  self				:	Union[str, Any]		,     __a				:	pa.StructArray							)			->  pa.StructArray:


												"""simple docstring"""

												@no_op_if_value_is_null
												def path_to_bytes(__a				:	Any							):
																			with xopen(__a		,     """rb"""							) as f:
																										__lowercase      :							List[Any]				=     f.read()
																			return bytes_

												__lowercase      :							Dict				=     pa.array(
												    [
												        (path_to_bytes(x["""path"""]							) if x["""bytes"""] is None else x["""bytes"""]) if x is not None else None
												        for x in storage.to_pylist()
												    ]		,     type=pa.binary()		,     )
												__lowercase      :							Dict				=     pa.array(
												    [os.path.basename(__a							) if path is not None else None for path in storage.field("""path"""							).to_pylist()]		,     type=pa.string()		,     )
												__lowercase      :							Optional[int]				=     pa.StructArray.from_arrays([bytes_array, path_array]		,     ["""bytes""", """path"""]		,     mask=bytes_array.is_null()							)
												return array_cast(__a		,     self.pa_type							)







def       snake_case_		(     ):
							if config.PIL_AVAILABLE:
														import PIL.Image
							else:
														raise ImportError("""To support encoding images, please install 'Pillow'."""      )

							global _IMAGE_COMPRESSION_FORMATS
							if _IMAGE_COMPRESSION_FORMATS is None:
														PIL.Image.init()
														__lowercase      :							List[str]				=     list(set(PIL.Image.OPEN.keys()      ) & set(PIL.Image.SAVE.keys()      )      )
							return _IMAGE_COMPRESSION_FORMATS
def       snake_case_		(     lowerCAmelCase_       :						"PIL.Image.Image"      ):
							__lowercase      :							List[Any]				=     BytesIO()
							if image.format in list_image_compression_formats():
														__lowercase      :							Any				=     image.format
							else:
														__lowercase      :							Any				=     """PNG""" if image.mode in ["""1""", """L""", """LA""", """RGB""", """RGBA"""] else """TIFF"""
							image.save(lowerCAmelCase_       ,       format=lowerCAmelCase_      )
							return buffer.getvalue()
def       snake_case_		(     lowerCAmelCase_       :						"PIL.Image.Image"      ):
							if hasattr(lowerCAmelCase_       ,       """filename"""      ) and image.filename != "":
														return {"path": image.filename, "bytes": None}
							else:
														return {"path": None, "bytes": image_to_bytes(lowerCAmelCase_      )}
def       snake_case_		(     lowerCAmelCase_       :						np.ndarray      ):
							if config.PIL_AVAILABLE:
														import PIL.Image
							else:
														raise ImportError("""To support encoding images, please install 'Pillow'."""      )

							__lowercase      :							List[str]				=     array.dtype
							__lowercase      :							Optional[Any]				=     dtype.byteorder if dtype.byteorder != """=""" else _NATIVE_BYTEORDER
							__lowercase      :							int				=     dtype.kind
							__lowercase      :							str				=     dtype.itemsize

							__lowercase      :							List[Any]				=     None

							# Multi-channel array case (only np.dtype("|u1") is allowed)
							if array.shape[2:]:
														__lowercase      :							Union[str, Any]				=     np.dtype("""|u1"""      )
														if dtype_kind not in ["u", "i"]:
																					raise TypeError(
																					    F"Unsupported array dtype {dtype} for image encoding. Only {dest_dtype} is supported for multi-channel arrays."      )
														if dtype is not dest_dtype:
																					warnings.warn(F"Downcasting array dtype {dtype} to {dest_dtype} to be compatible with 'Pillow'"      )
    # Exact match
							elif dtype in _VALID_IMAGE_ARRAY_DTPYES:
														__lowercase      :							int				=     dtype
							else:  # Downcast the type within the kind (np.can_cast(from_type, to_type, casting="same_kind") doesn't behave as expected, so do it manually)
														while dtype_itemsize >= 1:
																					__lowercase      :							Dict				=     dtype_byteorder + dtype_kind + str(lowerCAmelCase_      )
																					__lowercase      :							Optional[int]				=     np.dtype(lowerCAmelCase_      )
																					if dest_dtype in _VALID_IMAGE_ARRAY_DTPYES:
																												warnings.warn(F"Downcasting array dtype {dtype} to {dest_dtype} to be compatible with 'Pillow'"      )
																												break
																					else:
																												dtype_itemsize //= 2

							if dest_dtype is None:
														raise TypeError(
														    F"Cannot convert dtype {dtype} to a valid image dtype. Valid image dtypes: {_VALID_IMAGE_ARRAY_DTPYES}"      )
							__lowercase      :							Any				=     PIL.Image.fromarray(array.astype(lowerCAmelCase_      )      )
							return {"path": None, "bytes": image_to_bytes(lowerCAmelCase_      )}



def       snake_case_		(     lowerCAmelCase_       :						Union[List[str], List[dict], List[np.ndarray], List["PIL.Image.Image"]]      ):
							if config.PIL_AVAILABLE:
														import PIL.Image
							else:
														raise ImportError("""To support encoding images, please install 'Pillow'."""      )

							if objs:
														__lowercase      :							List[str]				=     first_non_null_value(lowerCAmelCase_      )
														if isinstance(lowerCAmelCase_       ,       lowerCAmelCase_      ):
																					return [{"path": obj, "bytes": None} if obj is not None else None for obj in objs]
														if isinstance(lowerCAmelCase_       ,       np.ndarray      ):
																					__lowercase      :							int				=     no_op_if_value_is_null(lowerCAmelCase_      )
																					return [obj_to_image_dict_func(lowerCAmelCase_      ) for obj in objs]
														elif isinstance(lowerCAmelCase_       ,       PIL.Image.Image      ):
																					__lowercase      :							str				=     no_op_if_value_is_null(lowerCAmelCase_      )
																					return [obj_to_image_dict_func(lowerCAmelCase_      ) for obj in objs]
														else:
																					return objs
							else:
														return objs







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 : Union[str, 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 		lowerCAmelCase     :

					'''simple docstring'''

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


												"""simple docstring"""

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

												__lowercase      :							str				=     context_length
												__lowercase      :							int				=     prediction_length + label_length
												__lowercase      :							Union[str, Any]				=     label_length

												__lowercase      :							Optional[int]				=     moving_average
												__lowercase      :							Optional[Any]				=     autocorrelation_factor



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


												"""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				:	Tuple		,     __a				:	str							)			->  int:


												"""simple docstring"""

												__lowercase      :							Any				=     config.context_length + max(config.lags_sequence							)

												__lowercase      :							Any				=     ids_tensor([self.batch_size, 1]		,     config.cardinality[0]							)
												__lowercase      :							Optional[int]				=     floats_tensor([self.batch_size, _past_length, config.num_time_features]							)
												__lowercase      :							List[str]				=     floats_tensor([self.batch_size, _past_length]							)
												__lowercase      :							List[str]				=     floats_tensor([self.batch_size, _past_length]							) > 0.5

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

												__lowercase      :							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       lowerCAmelCase				(  self				:	Union[str, Any]							)			->  Optional[Any]:


												"""simple docstring"""

												__lowercase      :							Optional[Any]				=     self.get_config()
												__lowercase      :							Any				=     self.prepare_autoformer_inputs_dict(__a							)
												return config, inputs_dict



					def       lowerCAmelCase				(  self				:	Optional[Any]							)			->  Tuple:


												"""simple docstring"""

												__lowercase			,			__lowercase      :							List[str]				=     self.prepare_config_and_inputs()
												return config, inputs_dict



					def       lowerCAmelCase				(  self				:	Optional[Any]		,     __a				:	Tuple		,     __a				:	Optional[int]							)			->  Any:


												"""simple docstring"""

												__lowercase      :							List[str]				=     AutoformerModel(config=__a							).to(__a							).eval()
												__lowercase      :							Optional[int]				=     model(**__a							)

												__lowercase      :							Dict				=     outputs.encoder_last_hidden_state
												__lowercase      :							Tuple				=     outputs.last_hidden_state

												with tempfile.TemporaryDirectory() as tmpdirname:
																			__lowercase      :							List[str]				=     model.get_encoder()
																			encoder.save_pretrained(__a							)
																			__lowercase      :							List[str]				=     AutoformerEncoder.from_pretrained(__a							).to(__a							)

												__lowercase			,			__lowercase			,			__lowercase			,			__lowercase			,			__lowercase      :							Any				=     model.create_network_inputs(**__a							)
												__lowercase			,			__lowercase      :							Any				=     model.decomposition_layer(transformer_inputs[:, : config.context_length, ...]							)

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

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

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

												__lowercase      :							str				=     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 		lowerCAmelCase     ( __a      ,       __a      ,       unittest.TestCase			):

					'''simple docstring'''

					_A						:			List[str]									=      (AutoformerModel, AutoformerForPrediction) if is_torch_available() else ()
					_A						:			List[Any]									=      (AutoformerForPrediction,) if is_torch_available() else ()
					_A						:			Any									=      {'''feature-extraction''': AutoformerModel} if is_torch_available() else {}
					_A						:			Dict									=      False
					_A						:			Tuple									=      False
					_A						:			Optional[int]									=      False
					_A						:			Tuple									=      False
					_A						:			str									=      False
					_A						:			Union[str, Any]									=      False



					def       lowerCAmelCase				(  self				:	Dict							)			->  str:


												"""simple docstring"""

												__lowercase      :							List[str]				=     AutoformerModelTester(self							)
												__lowercase      :							Dict				=     ConfigTester(self		,     config_class=__a		,     has_text_modality=__a							)



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


												"""simple docstring"""

												self.config_tester.run_common_tests()



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


												"""simple docstring"""

												__lowercase			,			__lowercase      :							Dict				=     self.model_tester.prepare_config_and_inputs()
												for model_class in self.all_model_classes:
																			__lowercase      :							Dict				=     model_class(__a							)

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



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


												"""simple docstring"""

												__lowercase      :							str				=     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				:	str							)			->  int:


												"""simple docstring"""

												pass



					def       lowerCAmelCase				(  self				:	Any							)			->  List[str]:


												"""simple docstring"""

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



					def       lowerCAmelCase				(  self				:	List[Any]							)			->  Optional[Any]:


												"""simple docstring"""

												__lowercase			,			__lowercase      :							Optional[Any]				=     self.model_tester.prepare_config_and_inputs_for_common()

												for model_class in self.all_model_classes:
																			__lowercase      :							Dict				=     model_class(__a							)
																			__lowercase      :							Union[str, Any]				=     inspect.signature(model.forward							)
																			# signature.parameters is an OrderedDict => so arg_names order is deterministic
																			__lowercase      :							Any				=     [*signature.parameters.keys()]

																			__lowercase      :							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(__a							)]		,     __a							)



					def       lowerCAmelCase				(  self				:	int							)			->  int:


												"""simple docstring"""

												__lowercase			,			__lowercase      :							str				=     self.model_tester.prepare_config_and_inputs_for_common()
												__lowercase      :							int				=     True

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

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

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

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

												__lowercase      :							Optional[Any]				=     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							)			->  Any:


												"""simple docstring"""

												super().test_retain_grad_hidden_states_attentions()







def       snake_case_		(     lowerCAmelCase_       :						Optional[int]="train-batch.pt"      ):
							__lowercase      :							Dict				=     hf_hub_download(repo_id="""hf-internal-testing/tourism-monthly-batch"""       ,       filename=lowerCAmelCase_       ,       repo_type="""dataset"""      )
							__lowercase      :							Optional[int]				=     torch.load(lowerCAmelCase_       ,       map_location=lowerCAmelCase_      )
							return batch



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

					'''simple docstring'''

					def       lowerCAmelCase				(  self				:	Dict							)			->  int:


												"""simple docstring"""

												__lowercase      :							List[str]				=     AutoformerModel.from_pretrained("""huggingface/autoformer-tourism-monthly"""							).to(__a							)
												__lowercase      :							List[Any]				=     prepare_batch()

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

												__lowercase      :							List[str]				=     torch.Size(
												    (64, model.config.prediction_length + model.config.label_length, model.config.feature_size)							)
												self.assertEqual(output.shape		,     __a							)

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



					def       lowerCAmelCase				(  self				:	str							)			->  str:


												"""simple docstring"""

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

												__lowercase      :							Optional[int]				=     torch.tensor(
												    [[-0.0734, -0.9036, 0.8358], [4.7186, 2.4113, 1.9581], [1.7953, 2.3558, 1.2970]]		,     device=__a							)
												self.assertTrue(torch.allclose(output[0, :3, :3]		,     __a		,     atol=__a							)							)



					def       lowerCAmelCase				(  self				:	Dict							)			->  List[Any]:


												"""simple docstring"""

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

												__lowercase      :							Optional[Any]				=     torch.tensor([3130.6763, 4056.5293, 7053.0786]		,     device=__a							)
												__lowercase      :							Dict				=     outputs.sequences.mean(dim=1							)
												self.assertTrue(torch.allclose(mean_prediction[0, -3:]		,     __a		,     rtol=1E-1							)							)