






'''simple docstring'''




from dataclasses import dataclass
from typing import 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 .modeling_utils import ModelMixin
from .vae import Decoder, DecoderOutput, Encoder, VectorQuantizer

@dataclass
class   _UpperCAmelCase     (	a__		):




   """simple docstring"""

   snake_case       							=		42





class   _UpperCAmelCase     (	a__				,       a__		):




   """simple docstring"""

   @register_to_config
   def __init__(						self					:      Optional[Any] , __UpperCAmelCase					:      Tuple = 3 , __UpperCAmelCase					:      Union[str, Any] = 3 , __UpperCAmelCase					:      Union[str, Any] = ("DownEncoderBlock2D",) , __UpperCAmelCase					:      int = ("UpDecoderBlock2D",) , __UpperCAmelCase					:      Optional[Any] = (64,) , __UpperCAmelCase					:      Tuple = 1 , __UpperCAmelCase					:      Dict = "silu" , __UpperCAmelCase					:      Dict = 3 , __UpperCAmelCase					:      str = 32 , __UpperCAmelCase					:      Union[str, Any] = 256 , __UpperCAmelCase					:      str = 32 , __UpperCAmelCase					:      List[Any] = None , __UpperCAmelCase					:      List[str] = 0.18215 , __UpperCAmelCase					:      str = "group" , ):





         '''simple docstring'''





         super().__init__()

         # pass init params to Encoder
         _A     	=      Encoder(
             in_channels=_UpperCAmelCase , out_channels=_UpperCAmelCase , down_block_types=_UpperCAmelCase , block_out_channels=_UpperCAmelCase , layers_per_block=_UpperCAmelCase , act_fn=_UpperCAmelCase , norm_num_groups=_UpperCAmelCase , double_z=_UpperCAmelCase , )

         _A     	=      vq_embed_dim if vq_embed_dim is not None else latent_channels

         _A     	=      nn.Convad(_UpperCAmelCase , _UpperCAmelCase , 1						)
         _A     	=      VectorQuantizer(_UpperCAmelCase , _UpperCAmelCase , beta=0.25 , remap=_UpperCAmelCase , sane_index_shape=_UpperCAmelCase						)
         _A     	=      nn.Convad(_UpperCAmelCase , _UpperCAmelCase , 1						)

         # pass init params to Decoder
         _A     	=      Decoder(
             in_channels=_UpperCAmelCase , out_channels=_UpperCAmelCase , up_block_types=_UpperCAmelCase , block_out_channels=_UpperCAmelCase , layers_per_block=_UpperCAmelCase , act_fn=_UpperCAmelCase , norm_num_groups=_UpperCAmelCase , norm_type=_UpperCAmelCase , )

   @apply_forward_hook
   def 		lowerCAmelCase					(						self					:      Dict , __UpperCAmelCase					:      List[Any] , __UpperCAmelCase					:      Union[str, Any] = True						):





         '''simple docstring'''





         _A     	=      self.encoder(_UpperCAmelCase						)
         _A     	=      self.quant_conv(_UpperCAmelCase						)

         if not return_dict:
               return (h,)

         return VQEncoderOutput(latents=_UpperCAmelCase						)

   @apply_forward_hook
   def 		lowerCAmelCase					(						self					:      Any , __UpperCAmelCase					:      Optional[int] , __UpperCAmelCase					:      List[str] = False , __UpperCAmelCase					:      List[str] = True						):





         '''simple docstring'''





         if not force_not_quantize:
               _A     	=      self.quantize(_UpperCAmelCase						)
         else:
               _A     	=      h
         _A     	=      self.post_quant_conv(_UpperCAmelCase						)
         _A     	=      self.decoder(_UpperCAmelCase , quant if self.config.norm_type == "spatial" else None						)

         if not return_dict:
               return (dec,)

         return DecoderOutput(sample=_UpperCAmelCase						)

   def 		lowerCAmelCase					(						self					:      Optional[int] , __UpperCAmelCase					:      List[Any] , __UpperCAmelCase					:      List[Any] = True						):





         '''simple docstring'''





         _A     	=      sample
         _A     	=      self.encode(_UpperCAmelCase						).latents
         _A     	=      self.decode(_UpperCAmelCase						).sample

         if not return_dict:
               return (dec,)

         return DecoderOutput(sample=_UpperCAmelCase						)












'''simple docstring'''




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

import torch
import torch.nn as nn

from ..configuration_utils import ConfigMixin, register_to_config
from ..utils import BaseOutput
from .embeddings import GaussianFourierProjection, TimestepEmbedding, Timesteps
from .modeling_utils import ModelMixin
from .unet_ad_blocks import get_down_block, get_mid_block, get_out_block, get_up_block


@dataclass
class 					SCREAMING_SNAKE_CASE       (a__				):
     lowerCAmelCase						     =			42




class 					SCREAMING_SNAKE_CASE       (a__   ,						a__				):


     @register_to_config
     def __init__(		self		,      _UpperCAmelCase = 6_5536		,      _UpperCAmelCase = None		,      _UpperCAmelCase = 2		,      _UpperCAmelCase = 2		,      _UpperCAmelCase = 0		,      _UpperCAmelCase = "fourier"		,      _UpperCAmelCase = True		,      _UpperCAmelCase = False		,      _UpperCAmelCase = 0.0		,      _UpperCAmelCase = ("DownBlock1DNoSkip", "DownBlock1D", "AttnDownBlock1D")		,      _UpperCAmelCase = ("AttnUpBlock1D", "UpBlock1D", "UpBlock1DNoSkip")		,      _UpperCAmelCase = "UNetMidBlock1D"		,      _UpperCAmelCase = None		,      _UpperCAmelCase = (32, 32, 64)		,      _UpperCAmelCase = None		,      _UpperCAmelCase = 8		,      _UpperCAmelCase = 1		,      _UpperCAmelCase = False		,      ):

          '''simple docstring'''






          super().__init__()
          __A		:					Dict						=			sample_size

          # time
          if time_embedding_type == "fourier":
               __A		:					int						=			GaussianFourierProjection(
                   embedding_size=8		,      set_W_to_weight=_UpperCAmelCase		,      log=_UpperCAmelCase		,      flip_sin_to_cos=_UpperCAmelCase)
               __A		:					Any						=			2 * block_out_channels[0]
          elif time_embedding_type == "positional":
               __A		:					List[str]						=			Timesteps(
                   block_out_channels[0]		,      flip_sin_to_cos=_UpperCAmelCase		,      downscale_freq_shift=_UpperCAmelCase)
               __A		:					List[str]						=			block_out_channels[0]

          if use_timestep_embedding:
               __A		:					Optional[Any]						=			block_out_channels[0] * 4
               __A		:					Optional[int]						=			TimestepEmbedding(
                   in_channels=_UpperCAmelCase		,      time_embed_dim=_UpperCAmelCase		,      act_fn=_UpperCAmelCase		,      out_dim=block_out_channels[0]		,      )

          __A		:					Dict						=			nn.ModuleList([])
          __A		:					Dict						=			None
          __A		:					Tuple						=			nn.ModuleList([])
          __A		:					Tuple						=			None

          # down
          __A		:					Any						=			in_channels
          for i, down_block_type in enumerate(_UpperCAmelCase):
               __A		:					Tuple						=			output_channel
               __A		:					Optional[Any]						=			block_out_channels[i]

               if i == 0:
                    input_channel += extra_in_channels

               __A		:					List[str]						=			i == len(_UpperCAmelCase) - 1

               __A		:					int						=			get_down_block(
                   _UpperCAmelCase		,      num_layers=_UpperCAmelCase		,      in_channels=_UpperCAmelCase		,      out_channels=_UpperCAmelCase		,      temb_channels=block_out_channels[0]		,      add_downsample=not is_final_block or downsample_each_block		,      )
               self.down_blocks.append(_UpperCAmelCase)

          # mid
          __A		:					str						=			get_mid_block(
              _UpperCAmelCase		,      in_channels=block_out_channels[-1]		,      mid_channels=block_out_channels[-1]		,      out_channels=block_out_channels[-1]		,      embed_dim=block_out_channels[0]		,      num_layers=_UpperCAmelCase		,      add_downsample=_UpperCAmelCase		,      )

          # up
          __A		:					Optional[int]						=			list(reversed(_UpperCAmelCase))
          __A		:					Optional[int]						=			reversed_block_out_channels[0]
          if out_block_type is None:
               __A		:					str						=			out_channels
          else:
               __A		:					List[Any]						=			block_out_channels[0]

          for i, up_block_type in enumerate(_UpperCAmelCase):
               __A		:					Optional[Any]						=			output_channel
               __A		:					Optional[Any]						=			(
                   reversed_block_out_channels[i + 1] if i < len(_UpperCAmelCase) - 1 else final_upsample_channels
               )

               __A		:					Dict						=			i == len(_UpperCAmelCase) - 1

               __A		:					str						=			get_up_block(
                   _UpperCAmelCase		,      num_layers=_UpperCAmelCase		,      in_channels=_UpperCAmelCase		,      out_channels=_UpperCAmelCase		,      temb_channels=block_out_channels[0]		,      add_upsample=not is_final_block		,      )
               self.up_blocks.append(_UpperCAmelCase)
               __A		:					Optional[int]						=			output_channel

          # out
          __A		:					str						=			norm_num_groups if norm_num_groups is not None else min(block_out_channels[0] // 4		,      32)
          __A		:					Optional[Any]						=			get_out_block(
              out_block_type=_UpperCAmelCase		,      num_groups_out=_UpperCAmelCase		,      embed_dim=block_out_channels[0]		,      out_channels=_UpperCAmelCase		,      act_fn=_UpperCAmelCase		,      fc_dim=block_out_channels[-1] // 4		,      )


     def 				SCREAMING_SNAKE_CASE    (		self		,      _UpperCAmelCase		,      _UpperCAmelCase		,      _UpperCAmelCase = True		,      ):

          '''simple docstring'''






          __A		:					Any						=			timestep
          if not torch.is_tensor(_UpperCAmelCase):
               __A		:					Any						=			torch.tensor([timesteps]		,      dtype=torch.long		,      device=sample.device)
          elif torch.is_tensor(_UpperCAmelCase) and len(timesteps.shape) == 0:
               __A		:					Any						=			timesteps[None].to(sample.device)

          __A		:					List[Any]						=			self.time_proj(_UpperCAmelCase)
          if self.config.use_timestep_embedding:
               __A		:					Dict						=			self.time_mlp(_UpperCAmelCase)
          else:
               __A		:					Dict						=			timestep_embed[..., None]
               __A		:					Tuple						=			timestep_embed.repeat([1, 1, sample.shape[2]]).to(sample.dtype)
               __A		:					List[Any]						=			timestep_embed.broadcast_to((sample.shape[:1] + timestep_embed.shape[1:]))

          # 2. down
          __A		:					int						=			()
          for downsample_block in self.down_blocks:
               __A					,__A		:					int						=			downsample_block(hidden_states=_UpperCAmelCase		,      temb=_UpperCAmelCase)
               down_block_res_samples += res_samples

          # 3. mid
          if self.mid_block:
               __A		:					Optional[int]						=			self.mid_block(_UpperCAmelCase		,      _UpperCAmelCase)

          # 4. up
          for i, upsample_block in enumerate(self.up_blocks):
               __A		:					Any						=			down_block_res_samples[-1:]
               __A		:					Optional[int]						=			down_block_res_samples[:-1]
               __A		:					Any						=			upsample_block(_UpperCAmelCase		,      res_hidden_states_tuple=_UpperCAmelCase		,      temb=_UpperCAmelCase)

          # 5. post-process
          if self.out_block:
               __A		:					Dict						=			self.out_block(_UpperCAmelCase		,      _UpperCAmelCase)

          if not return_dict:
               return (sample,)

          return UNetaDOutput(sample=_UpperCAmelCase)