

'''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 UNetMidBlockaD, get_down_block, get_up_block
@dataclass
class 	SCREAMING_SNAKE_CASE__			(				_UpperCAmelCase      ):
    _A			    =							42






class 	SCREAMING_SNAKE_CASE__			(				_UpperCAmelCase,_UpperCAmelCase      ):




    @register_to_config
    def __init__(					self				,     lowercase__ = None				,     lowercase__ = 3				,     lowercase__ = 3				,     lowercase__ = False				,     lowercase__ = "positional"				,     lowercase__ = 0				,     lowercase__ = True				,     lowercase__ = ("DownBlock2D", "AttnDownBlock2D", "AttnDownBlock2D", "AttnDownBlock2D")				,     lowercase__ = ("AttnUpBlock2D", "AttnUpBlock2D", "AttnUpBlock2D", "UpBlock2D")				,     lowercase__ = (224, 448, 672, 896)				,     lowercase__ = 2				,     lowercase__ = 1				,     lowercase__ = 1				,     lowercase__ = "conv"				,     lowercase__ = "conv"				,     lowercase__ = "silu"				,     lowercase__ = 8				,     lowercase__ = 32				,     lowercase__ = 1e-5				,     lowercase__ = "default"				,     lowercase__ = True				,     lowercase__ = None				,     lowercase__ = None				,     ):






          """simple docstring"""






          super().__init__()

          SCREAMING_SNAKE_CASE_      :			str     =       sample_size
          SCREAMING_SNAKE_CASE_      :			Dict     =       block_out_channels[0] * 4

          # Check inputs
          if len(lowercase__			) != len(lowercase__			):
                raise ValueError(
                    F"Must provide the same number of `down_block_types` as `up_block_types`. `down_block_types`: {down_block_types}. `up_block_types`: {up_block_types}."			)

          if len(lowercase__			) != len(lowercase__			):
                raise ValueError(
                    F"Must provide the same number of `block_out_channels` as `down_block_types`. `block_out_channels`: {block_out_channels}. `down_block_types`: {down_block_types}."			)

          # input
          SCREAMING_SNAKE_CASE_      :			Tuple     =       nn.Convad(lowercase__				,     block_out_channels[0]				,     kernel_size=3				,     padding=(1, 1)			)

          # time
          if time_embedding_type == "fourier":
                SCREAMING_SNAKE_CASE_      :			Tuple     =       GaussianFourierProjection(embedding_size=block_out_channels[0]				,     scale=16			)
                SCREAMING_SNAKE_CASE_      :			Tuple     =       2 * block_out_channels[0]
          elif time_embedding_type == "positional":
                SCREAMING_SNAKE_CASE_      :			Any     =       Timesteps(block_out_channels[0]				,     lowercase__				,     lowercase__			)
                SCREAMING_SNAKE_CASE_      :			List[str]     =       block_out_channels[0]

          SCREAMING_SNAKE_CASE_      :			Dict     =       TimestepEmbedding(lowercase__				,     lowercase__			)

          # class embedding
          if class_embed_type is None and num_class_embeds is not None:
                SCREAMING_SNAKE_CASE_      :			int     =       nn.Embedding(lowercase__				,     lowercase__			)
          elif class_embed_type == "timestep":
                SCREAMING_SNAKE_CASE_      :			Tuple     =       TimestepEmbedding(lowercase__				,     lowercase__			)
          elif class_embed_type == "identity":
                SCREAMING_SNAKE_CASE_      :			Dict     =       nn.Identity(lowercase__				,     lowercase__			)
          else:
                SCREAMING_SNAKE_CASE_      :			str     =       None

          SCREAMING_SNAKE_CASE_      :			Optional[Any]     =       nn.ModuleList([]			)
          SCREAMING_SNAKE_CASE_      :			Dict     =       None
          SCREAMING_SNAKE_CASE_      :			Optional[Any]     =       nn.ModuleList([]			)

          # down
          SCREAMING_SNAKE_CASE_      :			List[str]     =       block_out_channels[0]
          for i, down_block_type in enumerate(lowercase__			):
                SCREAMING_SNAKE_CASE_      :			Tuple     =       output_channel
                SCREAMING_SNAKE_CASE_      :			Tuple     =       block_out_channels[i]
                SCREAMING_SNAKE_CASE_      :			Tuple     =       i == len(lowercase__			) - 1

                SCREAMING_SNAKE_CASE_      :			List[str]     =       get_down_block(
                    lowercase__				,     num_layers=lowercase__				,     in_channels=lowercase__				,     out_channels=lowercase__				,     temb_channels=lowercase__				,     add_downsample=not is_final_block				,     resnet_eps=lowercase__				,     resnet_act_fn=lowercase__				,     resnet_groups=lowercase__				,     attention_head_dim=attention_head_dim if attention_head_dim is not None else output_channel				,     downsample_padding=lowercase__				,     resnet_time_scale_shift=lowercase__				,     downsample_type=lowercase__				,     )
                self.down_blocks.append(lowercase__			)

          # mid
          SCREAMING_SNAKE_CASE_      :			int     =       UNetMidBlockaD(
              in_channels=block_out_channels[-1]				,     temb_channels=lowercase__				,     resnet_eps=lowercase__				,     resnet_act_fn=lowercase__				,     output_scale_factor=lowercase__				,     resnet_time_scale_shift=lowercase__				,     attention_head_dim=attention_head_dim if attention_head_dim is not None else block_out_channels[-1]				,     resnet_groups=lowercase__				,     add_attention=lowercase__				,     )

          # up
          SCREAMING_SNAKE_CASE_      :			List[str]     =       list(reversed(lowercase__			)			)
          SCREAMING_SNAKE_CASE_      :			Union[str, Any]     =       reversed_block_out_channels[0]
          for i, up_block_type in enumerate(lowercase__			):
                SCREAMING_SNAKE_CASE_      :			Optional[Any]     =       output_channel
                SCREAMING_SNAKE_CASE_      :			int     =       reversed_block_out_channels[i]
                SCREAMING_SNAKE_CASE_      :			Dict     =       reversed_block_out_channels[min(i + 1				,     len(lowercase__			) - 1			)]

                SCREAMING_SNAKE_CASE_      :			str     =       i == len(lowercase__			) - 1

                SCREAMING_SNAKE_CASE_      :			Optional[int]     =       get_up_block(
                    lowercase__				,     num_layers=layers_per_block + 1				,     in_channels=lowercase__				,     out_channels=lowercase__				,     prev_output_channel=lowercase__				,     temb_channels=lowercase__				,     add_upsample=not is_final_block				,     resnet_eps=lowercase__				,     resnet_act_fn=lowercase__				,     resnet_groups=lowercase__				,     attention_head_dim=attention_head_dim if attention_head_dim is not None else output_channel				,     resnet_time_scale_shift=lowercase__				,     upsample_type=lowercase__				,     )
                self.up_blocks.append(lowercase__			)
                SCREAMING_SNAKE_CASE_      :			str     =       output_channel

          # out
          SCREAMING_SNAKE_CASE_      :			str     =       norm_num_groups if norm_num_groups is not None else min(block_out_channels[0] // 4				,     32			)
          SCREAMING_SNAKE_CASE_      :			List[str]     =       nn.GroupNorm(num_channels=block_out_channels[0]				,     num_groups=lowercase__				,     eps=lowercase__			)
          SCREAMING_SNAKE_CASE_      :			Tuple     =       nn.SiLU()
          SCREAMING_SNAKE_CASE_      :			str     =       nn.Convad(block_out_channels[0]				,     lowercase__				,     kernel_size=3				,     padding=1			)







    def   __lowerCamelCase      (					self				,     lowercase__				,     lowercase__				,     lowercase__ = None				,     lowercase__ = True				,     ):






          """simple docstring"""






          if self.config.center_input_sample:
                SCREAMING_SNAKE_CASE_      :			Dict     =       2 * sample - 1.0

          # 1. time
          SCREAMING_SNAKE_CASE_      :			Union[str, Any]     =       timestep
          if not torch.is_tensor(lowercase__			):
                SCREAMING_SNAKE_CASE_      :			Optional[int]     =       torch.tensor([timesteps]				,     dtype=torch.long				,     device=sample.device			)
          elif torch.is_tensor(lowercase__			) and len(timesteps.shape			) == 0:
                SCREAMING_SNAKE_CASE_      :			int     =       timesteps[None].to(sample.device			)

          # broadcast to batch dimension in a way that's compatible with ONNX/Core ML
          SCREAMING_SNAKE_CASE_      :			Union[str, Any]     =       timesteps * torch.ones(sample.shape[0]				,     dtype=timesteps.dtype				,     device=timesteps.device			)

          SCREAMING_SNAKE_CASE_      :			Optional[Any]     =       self.time_proj(lowercase__			)

          # timesteps does not contain any weights and will always return f32 tensors
          # but time_embedding might actually be running in fp16. so we need to cast here.
          # there might be better ways to encapsulate this.
          SCREAMING_SNAKE_CASE_      :			List[str]     =       t_emb.to(dtype=self.dtype			)
          SCREAMING_SNAKE_CASE_      :			Tuple     =       self.time_embedding(lowercase__			)

          if self.class_embedding is not None:
                if class_labels is None:
                      raise ValueError("class_labels should be provided when doing class conditioning"			)

                if self.config.class_embed_type == "timestep":
                      SCREAMING_SNAKE_CASE_      :			Dict     =       self.time_proj(lowercase__			)

                SCREAMING_SNAKE_CASE_      :			int     =       self.class_embedding(lowercase__			).to(dtype=self.dtype			)
                SCREAMING_SNAKE_CASE_      :			Optional[Any]     =       emb + class_emb

          # 2. pre-process
          SCREAMING_SNAKE_CASE_      :			str     =       sample
          SCREAMING_SNAKE_CASE_      :			Optional[Any]     =       self.conv_in(lowercase__			)

          # 3. down
          SCREAMING_SNAKE_CASE_      :			Optional[int]     =       (sample,)
          for downsample_block in self.down_blocks:
                if hasattr(lowercase__				,     "skip_conv"			):
                      SCREAMING_SNAKE_CASE_,						SCREAMING_SNAKE_CASE_,						SCREAMING_SNAKE_CASE_      :			Any     =       downsample_block(
                          hidden_states=lowercase__				,     temb=lowercase__				,     skip_sample=lowercase__			)
                else:
                      SCREAMING_SNAKE_CASE_,						SCREAMING_SNAKE_CASE_      :			int     =       downsample_block(hidden_states=lowercase__				,     temb=lowercase__			)

                down_block_res_samples += res_samples

          # 4. mid
          SCREAMING_SNAKE_CASE_      :			Dict     =       self.mid_block(lowercase__				,     lowercase__			)

          # 5. up
          SCREAMING_SNAKE_CASE_      :			List[Any]     =       None
          for upsample_block in self.up_blocks:
                SCREAMING_SNAKE_CASE_      :			Tuple     =       down_block_res_samples[-len(upsample_block.resnets			) :]
                SCREAMING_SNAKE_CASE_      :			Any     =       down_block_res_samples[: -len(upsample_block.resnets			)]

                if hasattr(lowercase__				,     "skip_conv"			):
                      SCREAMING_SNAKE_CASE_,						SCREAMING_SNAKE_CASE_      :			int     =       upsample_block(lowercase__				,     lowercase__				,     lowercase__				,     lowercase__			)
                else:
                      SCREAMING_SNAKE_CASE_      :			Union[str, Any]     =       upsample_block(lowercase__				,     lowercase__				,     lowercase__			)

        # 6. post-process
          SCREAMING_SNAKE_CASE_      :			str     =       self.conv_norm_out(lowercase__			)
          SCREAMING_SNAKE_CASE_      :			Dict     =       self.conv_act(lowercase__			)
          SCREAMING_SNAKE_CASE_      :			Dict     =       self.conv_out(lowercase__			)

          if skip_sample is not None:
                sample += skip_sample

          if self.config.time_embedding_type == "fourier":
                SCREAMING_SNAKE_CASE_      :			Optional[int]     =       timesteps.reshape((sample.shape[0], *([1] * len(sample.shape[1:]			)))			)
                SCREAMING_SNAKE_CASE_      :			int     =       sample / timesteps

          if not return_dict:
                return (sample,)

          return UNetaDOutput(sample=lowercase__			)




'''simple docstring'''






from ..utils import DummyObject, requires_backends
class 	SCREAMING_SNAKE_CASE__			(				metaclass=_UpperCAmelCase      ):
    _A			    =							["torch", "transformers", "invisible_watermark"]




    def __init__(					self				,     *lowercase__				,     **lowercase__			):






          """simple docstring"""






          requires_backends(self				,     ["torch", "transformers", "invisible_watermark"]			)




    @classmethod
    def   __lowerCamelCase      (					cls				,     *lowercase__				,     **lowercase__			):






          """simple docstring"""






          requires_backends(cls				,     ["torch", "transformers", "invisible_watermark"]			)







    @classmethod
    def   __lowerCamelCase      (					cls				,     *lowercase__				,     **lowercase__			):






          """simple docstring"""






          requires_backends(cls				,     ["torch", "transformers", "invisible_watermark"]			)






class 	SCREAMING_SNAKE_CASE__			(				metaclass=_UpperCAmelCase      ):
    _A			    =							["torch", "transformers", "invisible_watermark"]




    def __init__(					self				,     *lowercase__				,     **lowercase__			):






          """simple docstring"""






          requires_backends(self				,     ["torch", "transformers", "invisible_watermark"]			)




    @classmethod
    def   __lowerCamelCase      (					cls				,     *lowercase__				,     **lowercase__			):






          """simple docstring"""






          requires_backends(cls				,     ["torch", "transformers", "invisible_watermark"]			)







    @classmethod
    def   __lowerCamelCase      (					cls				,     *lowercase__				,     **lowercase__			):






          """simple docstring"""






          requires_backends(cls				,     ["torch", "transformers", "invisible_watermark"]			)

