
'''simple docstring'''




from dataclasses import dataclass
from typing import Dict, Optional, Tuple, Union

import torch
import torch.nn as nn

from ..configuration_utils import ConfigMixin, register_to_config
from ..utils import BaseOutput, apply_forward_hook
from .attention_processor import AttentionProcessor, AttnProcessor
from .modeling_utils import ModelMixin
from .vae import Decoder, DecoderOutput, DiagonalGaussianDistribution, Encoder





@dataclass
class 					A							(				a__		):
     __magic_name__          =						42





class 					A							(				a__       ,     a__		):
     __magic_name__          =						True





     @register_to_config
     def __init__(				self		,     SCREAMING_SNAKE_CASE = 3		,     SCREAMING_SNAKE_CASE = 3		,     SCREAMING_SNAKE_CASE = ("DownEncoderBlock2D",)		,     SCREAMING_SNAKE_CASE = ("UpDecoderBlock2D",)		,     SCREAMING_SNAKE_CASE = (64,)		,     SCREAMING_SNAKE_CASE = 1		,     SCREAMING_SNAKE_CASE = "silu"		,     SCREAMING_SNAKE_CASE = 4		,     SCREAMING_SNAKE_CASE = 32		,     SCREAMING_SNAKE_CASE = 32		,     SCREAMING_SNAKE_CASE = 0.18_215		,     )					->							Optional[Any]:






         """simple docstring"""






         super().__init__()

         # pass init params to Encoder
         A			:  Any   =		Encoder(
             in_channels=__a		,     out_channels=__a		,     down_block_types=__a		,     block_out_channels=__a		,     layers_per_block=__a		,     act_fn=__a		,     norm_num_groups=__a		,     double_z=__a		,     )

         # pass init params to Decoder
         A			:  str   =		Decoder(
             in_channels=__a		,     out_channels=__a		,     up_block_types=__a		,     block_out_channels=__a		,     layers_per_block=__a		,     norm_num_groups=__a		,     act_fn=__a		,     )

         A			:  Any   =		nn.Convad(2 * latent_channels		,     2 * latent_channels		,     1      )
         A			:  Dict   =		nn.Convad(__a		,     __a		,     1      )

         A			:  str   =		False
         A			:  List[str]   =		False

         # only relevant if vae tiling is enabled
         A			:  Tuple   =		self.config.sample_size
         A			:  int   =		(
             self.config.sample_size[0]
             if isinstance(self.config.sample_size		,     (list, tuple)      )
             else self.config.sample_size
         )
         A			:  Union[str, Any]   =		int(sample_size / (2 ** (len(self.config.block_out_channels      ) - 1))      )
         A			:  List[Any]   =		0.25





     def 							__lowerCAmelCase				(				self		,     SCREAMING_SNAKE_CASE		,     SCREAMING_SNAKE_CASE=False      )					->							List[str]:






         """simple docstring"""






         if isinstance(__a		,     (Encoder, Decoder)      ):
             A			:  int   =		value





     def 							__lowerCAmelCase				(				self		,     SCREAMING_SNAKE_CASE = True      )					->							str:






         """simple docstring"""






         A			:  Optional[Any]   =		use_tiling





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






         """simple docstring"""






         self.enable_tiling(__a      )





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






         """simple docstring"""






         A			:  Tuple   =		True





     def 							__lowerCAmelCase				(				self      )					->							Dict:






         """simple docstring"""






         A			:  Tuple   =		False





     @property
     # Copied from diffusers.models.unet_2d_condition.UNet2DConditionModel.attn_processors
     def 							__lowerCAmelCase				(				self      )					->							Dict[str, AttentionProcessor]:






         """simple docstring"""






         A			:  Tuple   =		{}

         def fn_recursive_add_processors(SCREAMING_SNAKE_CASE		,     SCREAMING_SNAKE_CASE		,     SCREAMING_SNAKE_CASE      ):
             if hasattr(__a		,     '''set_processor'''      ):
                 A			:  Tuple   =		module.processor

             for sub_name, child in module.named_children():
                 fn_recursive_add_processors(F'{name}.{sub_name}'		,     __a		,     __a      )

             return processors

         for name, module in self.named_children():
             fn_recursive_add_processors(__a		,     __a		,     __a      )

         return processors





     def 							__lowerCAmelCase				(				self		,     SCREAMING_SNAKE_CASE      )					->							Optional[int]:






         """simple docstring"""






         A			:  Any   =		len(self.attn_processors.keys()      )

         if isinstance(__a		,     __a      ) and len(__a      ) != count:
             raise ValueError(
                 F'A dict of processors was passed, but the number of processors {len(__a      )} does not match the'
                 F' number of attention layers: {count}. Please make sure to pass {count} processor classes.'      )

         def fn_recursive_attn_processor(SCREAMING_SNAKE_CASE		,     SCREAMING_SNAKE_CASE		,     SCREAMING_SNAKE_CASE      ):
             if hasattr(__a		,     '''set_processor'''      ):
                 if not isinstance(__a		,     __a      ):
                     module.set_processor(__a      )
                 else:
                     module.set_processor(processor.pop(F'{name}.processor'      )      )

             for sub_name, child in module.named_children():
                 fn_recursive_attn_processor(F'{name}.{sub_name}'		,     __a		,     __a      )

         for name, module in self.named_children():
             fn_recursive_attn_processor(__a		,     __a		,     __a      )





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






         """simple docstring"""






         self.set_attn_processor(AttnProcessor()      )





     @apply_forward_hook
     def 							__lowerCAmelCase				(				self		,     SCREAMING_SNAKE_CASE		,     SCREAMING_SNAKE_CASE = True      )					->							AutoencoderKLOutput:






         """simple docstring"""






         if self.use_tiling and (x.shape[-1] > self.tile_sample_min_size or x.shape[-2] > self.tile_sample_min_size):
             return self.tiled_encode(__a		,     return_dict=__a      )

         if self.use_slicing and x.shape[0] > 1:
             A			:  int   =		[self.encoder(__a      ) for x_slice in x.split(1      )]
             A			:  Dict   =		torch.cat(__a      )
         else:
             A			:  Dict   =		self.encoder(__a      )

         A			:  List[Any]   =		self.quant_conv(__a      )
         A			:  Optional[int]   =		DiagonalGaussianDistribution(__a      )

         if not return_dict:
             return (posterior,)

         return AutoencoderKLOutput(latent_dist=__a      )





     def 							__lowerCAmelCase				(				self		,     SCREAMING_SNAKE_CASE		,     SCREAMING_SNAKE_CASE = True      )					->							Union[DecoderOutput, torch.FloatTensor]:






         """simple docstring"""






         if self.use_tiling and (z.shape[-1] > self.tile_latent_min_size or z.shape[-2] > self.tile_latent_min_size):
             return self.tiled_decode(__a		,     return_dict=__a      )

         A			:  Optional[Any]   =		self.post_quant_conv(__a      )
         A			:  Union[str, Any]   =		self.decoder(__a      )

         if not return_dict:
             return (dec,)

         return DecoderOutput(sample=__a      )





     @apply_forward_hook
     def 							__lowerCAmelCase				(				self		,     SCREAMING_SNAKE_CASE		,     SCREAMING_SNAKE_CASE = True      )					->							Union[DecoderOutput, torch.FloatTensor]:






         """simple docstring"""






         if self.use_slicing and z.shape[0] > 1:
             A			:  Any   =		[self._decode(__a      ).sample for z_slice in z.split(1      )]
             A			:  Optional[int]   =		torch.cat(__a      )
         else:
             A			:  Optional[int]   =		self._decode(__a      ).sample

         if not return_dict:
             return (decoded,)

         return DecoderOutput(sample=__a      )





     def 							__lowerCAmelCase				(				self		,     SCREAMING_SNAKE_CASE		,     SCREAMING_SNAKE_CASE		,     SCREAMING_SNAKE_CASE      )					->							int:






         """simple docstring"""






         A			:  str   =		min(a.shape[2]		,     b.shape[2]		,     __a      )
         for y in range(__a      ):
             A			:  int   =		a[:, :, -blend_extent + y, :] * (1 - y / blend_extent) + b[:, :, y, :] * (y / blend_extent)
         return b





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






         """simple docstring"""






         A			:  List[Any]   =		min(a.shape[3]		,     b.shape[3]		,     __a      )
         for x in range(__a      ):
             A			:  Tuple   =		a[:, :, :, -blend_extent + x] * (1 - x / blend_extent) + b[:, :, :, x] * (x / blend_extent)
         return b





     def 							__lowerCAmelCase				(				self		,     SCREAMING_SNAKE_CASE		,     SCREAMING_SNAKE_CASE = True      )					->							AutoencoderKLOutput:






         """simple docstring"""






         A			:  List[str]   =		int(self.tile_sample_min_size * (1 - self.tile_overlap_factor)      )
         A			:  Dict   =		int(self.tile_latent_min_size * self.tile_overlap_factor      )
         A			:  List[Any]   =		self.tile_latent_min_size - blend_extent

         # Split the image into 512x512 tiles and encode them separately.
         A			:  List[str]   =		[]
         for i in range(0		,     x.shape[2]		,     __a      ):
             A			:  Tuple   =		[]
             for j in range(0		,     x.shape[3]		,     __a      ):
                 A			:  List[Any]   =		x[:, :, i : i + self.tile_sample_min_size, j : j + self.tile_sample_min_size]
                 A			:  List[str]   =		self.encoder(__a      )
                 A			:  str   =		self.quant_conv(__a      )
                 row.append(__a      )
             rows.append(__a      )
         A			:  Tuple   =		[]
         for i, row in enumerate(__a      ):
             A			:  Union[str, Any]   =		[]
             for j, tile in enumerate(__a      ):
                 # blend the above tile and the left tile
                 # to the current tile and add the current tile to the result row
                 if i > 0:
                     A			:  Tuple   =		self.blend_v(rows[i - 1][j]		,     __a		,     __a      )
                 if j > 0:
                     A			:  int   =		self.blend_h(row[j - 1]		,     __a		,     __a      )
                 result_row.append(tile[:, :, :row_limit, :row_limit]      )
             result_rows.append(torch.cat(__a		,     dim=3      )      )

         A			:  str   =		torch.cat(__a		,     dim=2      )
         A			:  str   =		DiagonalGaussianDistribution(__a      )

         if not return_dict:
             return (posterior,)

         return AutoencoderKLOutput(latent_dist=__a      )





     def 							__lowerCAmelCase				(				self		,     SCREAMING_SNAKE_CASE		,     SCREAMING_SNAKE_CASE = True      )					->							Union[DecoderOutput, torch.FloatTensor]:






         """simple docstring"""






         A			:  Dict   =		int(self.tile_latent_min_size * (1 - self.tile_overlap_factor)      )
         A			:  str   =		int(self.tile_sample_min_size * self.tile_overlap_factor      )
         A			:  Any   =		self.tile_sample_min_size - blend_extent

         # Split z into overlapping 64x64 tiles and decode them separately.
         # The tiles have an overlap to avoid seams between tiles.
         A			:  Union[str, Any]   =		[]
         for i in range(0		,     z.shape[2]		,     __a      ):
             A			:  Optional[int]   =		[]
             for j in range(0		,     z.shape[3]		,     __a      ):
                 A			:  Tuple   =		z[:, :, i : i + self.tile_latent_min_size, j : j + self.tile_latent_min_size]
                 A			:  List[str]   =		self.post_quant_conv(__a      )
                 A			:  Any   =		self.decoder(__a      )
                 row.append(__a      )
             rows.append(__a      )
         A			:  Optional[Any]   =		[]
         for i, row in enumerate(__a      ):
             A			:  Union[str, Any]   =		[]
             for j, tile in enumerate(__a      ):
                 # blend the above tile and the left tile
                 # to the current tile and add the current tile to the result row
                 if i > 0:
                     A			:  Any   =		self.blend_v(rows[i - 1][j]		,     __a		,     __a      )
                 if j > 0:
                     A			:  List[Any]   =		self.blend_h(row[j - 1]		,     __a		,     __a      )
                 result_row.append(tile[:, :, :row_limit, :row_limit]      )
             result_rows.append(torch.cat(__a		,     dim=3      )      )

         A			:  int   =		torch.cat(__a		,     dim=2      )
         if not return_dict:
             return (dec,)

         return DecoderOutput(sample=__a      )





     def 							__lowerCAmelCase				(				self		,     SCREAMING_SNAKE_CASE		,     SCREAMING_SNAKE_CASE = False		,     SCREAMING_SNAKE_CASE = True		,     SCREAMING_SNAKE_CASE = None		,     )					->							Union[DecoderOutput, torch.FloatTensor]:






         """simple docstring"""






         A			:  List[str]   =		sample
         A			:  str   =		self.encode(__a      ).latent_dist
         if sample_posterior:
             A			:  int   =		posterior.sample(generator=__a      )
         else:
             A			:  str   =		posterior.mode()
         A			:  str   =		self.decode(__a      ).sample

         if not return_dict:
             return (dec,)

         return DecoderOutput(sample=__a      )










'''simple docstring'''







from typing import List, Optional

from ...configuration_utils import PretrainedConfig
from ...utils import logging


lowercase      :   Optional[Any]      =       logging.get_logger(__name__)

lowercase      :   Any      =       {
    """huggingface/autoformer-tourism-monthly""": """https://huggingface.co/huggingface/autoformer-tourism-monthly/resolve/main/config.json""",
}







class 					_a (a__	):





     '''simple docstring'''



     lowerCAmelCase_  :     List[str]          =					"""autoformer"""
     lowerCAmelCase_  :     int          =					{
         """hidden_size""": """d_model""",
         """num_attention_heads""": """encoder_attention_heads""",
         """num_hidden_layers""": """encoder_layers""",
     }







     def __init__(				self							,__a = None							,__a = None							,__a = "student_t"							,__a = "nll"							,__a = 1							,__a = [1, 2, 3, 4, 5, 6, 7]							,__a = True							,__a = 0							,__a = 0							,__a = 0							,__a = 0							,__a = None							,__a = None							,__a = 64							,__a = 2							,__a = 2							,__a = 2							,__a = 2							,__a = 32							,__a = 32							,__a = "gelu"							,__a = 0.1							,__a = 0.1							,__a = 0.1							,__a = 0.1							,__a = 0.1							,__a = 100							,__a = 0.02							,__a = True							,__a=True							,__a = 10							,__a = 25							,__a = 3							,**__a							,)      ->				Union[str, Any]:
          # time series specific configuration
          snake_case			:      List[str]        =			prediction_length
          snake_case			:      str        =			context_length if context_length is not None else prediction_length
          snake_case			:      List[Any]        =			distribution_output
          snake_case			:      Any        =			loss
          snake_case			:      Union[str, Any]        =			input_size
          snake_case			:      Optional[int]        =			num_time_features
          snake_case			:      Union[str, Any]        =			lags_sequence
          snake_case			:      Any        =			scaling
          snake_case			:      str        =			num_dynamic_real_features
          snake_case			:      Union[str, Any]        =			num_static_real_features
          snake_case			:      Optional[int]        =			num_static_categorical_features
          if cardinality is not None and num_static_categorical_features > 0:
               if len(__a							) != num_static_categorical_features:
                    raise ValueError(
                        """The cardinality should be a list of the same length as `num_static_categorical_features`"""							)
               snake_case			:      int        =			cardinality
          else:
               snake_case			:      Optional[Any]        =			[0]
          if embedding_dimension is not None and num_static_categorical_features > 0:
               if len(__a							) != num_static_categorical_features:
                    raise ValueError(
                        """The embedding dimension should be a list of the same length as `num_static_categorical_features`"""							)
               snake_case			:      int        =			embedding_dimension
          else:
               snake_case			:      List[str]        =			[min(50							,(cat + 1) // 2							) for cat in self.cardinality]
          snake_case			:      Optional[int]        =			num_parallel_samples

          # Transformer architecture configuration
          snake_case			:      Any        =			input_size * len(self.lags_sequence							) + self._number_of_features
          snake_case			:      Union[str, Any]        =			d_model
          snake_case			:      Any        =			encoder_attention_heads
          snake_case			:      Union[str, Any]        =			decoder_attention_heads
          snake_case			:      Optional[Any]        =			encoder_ffn_dim
          snake_case			:      Any        =			decoder_ffn_dim
          snake_case			:      List[Any]        =			encoder_layers
          snake_case			:      List[str]        =			decoder_layers

          snake_case			:      str        =			dropout
          snake_case			:      Union[str, Any]        =			attention_dropout
          snake_case			:      Optional[Any]        =			activation_dropout
          snake_case			:      Tuple        =			encoder_layerdrop
          snake_case			:      Any        =			decoder_layerdrop

          snake_case			:      Any        =			activation_function
          snake_case			:      int        =			init_std

          snake_case			:      Optional[Any]        =			use_cache

          # Autoformer
          snake_case			:      int        =			label_length
          snake_case			:      str        =			moving_average
          snake_case			:      Tuple        =			autocorrelation_factor

          super().__init__(is_encoder_decoder=__a							,**__a							)







     @property
     def 			snake_case_      (				self							)      ->				int:
          return (
              sum(self.embedding_dimension							)
              + self.num_dynamic_real_features
              + self.num_time_features
              + self.num_static_real_features
              + self.input_size * 2  # the log1p(abs(loc)) and log(scale) features
          )



