

from typing import Any, Dict, Optional

import torch
import torch.nn.functional as F
from torch import nn

from ..utils import maybe_allow_in_graph
from .activations import get_activation
from .attention_processor import Attention
from .embeddings import CombinedTimestepLabelEmbeddings


@maybe_allow_in_graph
class SCREAMING_SNAKE_CASE  (		nn.Module							):






			"""simple docstring"""




			def __init__(    self ,				__UpperCamelCase ,				__UpperCamelCase ,				__UpperCamelCase ,				__UpperCamelCase=0.0 ,				__UpperCamelCase = None ,				__UpperCamelCase = "geglu" ,				__UpperCamelCase = None ,				__UpperCamelCase = False ,				__UpperCamelCase = False ,				__UpperCamelCase = False ,				__UpperCamelCase = False ,				__UpperCamelCase = True ,				__UpperCamelCase = "layer_norm" ,				__UpperCamelCase = False ,				):

				"""simple docstring"""


				super().__init__()
				snake_case_      		= only_cross_attention

				snake_case_      		= (num_embeds_ada_norm is not None) and norm_type == 'ada_norm_zero'
				snake_case_      		= (num_embeds_ada_norm is not None) and norm_type == 'ada_norm'

				if norm_type in ("ada_norm", "ada_norm_zero") and num_embeds_ada_norm is None:
					raise ValueError(
					    f"""`norm_type` is set to {norm_type}, but `num_embeds_ada_norm` is not defined. Please make sure to"""
					    f""" define `num_embeds_ada_norm` if setting `norm_type` to {norm_type}."""							)

				# Define 3 blocks. Each block has its own normalization layer.
				# 1. Self-Attn
				if self.use_ada_layer_norm:
					snake_case_      		= AdaLayerNorm(__UpperCamelCase ,				__UpperCamelCase							)
				elif self.use_ada_layer_norm_zero:
					snake_case_      		= AdaLayerNormZero(__UpperCamelCase ,				__UpperCamelCase							)
				else:
					snake_case_      		= nn.LayerNorm(__UpperCamelCase ,				elementwise_affine=__UpperCamelCase							)
				snake_case_      		= Attention(
				    query_dim=__UpperCamelCase ,				heads=__UpperCamelCase ,				dim_head=__UpperCamelCase ,				dropout=__UpperCamelCase ,				bias=__UpperCamelCase ,				cross_attention_dim=cross_attention_dim if only_cross_attention else None ,				upcast_attention=__UpperCamelCase ,				)

				# 2. Cross-Attn
				if cross_attention_dim is not None or double_self_attention:
					# We currently only use AdaLayerNormZero for self attention where there will only be one attention block.
					# I.e. the number of returned modulation chunks from AdaLayerZero would not make sense if returned during
					# the second cross attention block.
					snake_case_      		= (
					    AdaLayerNorm(__UpperCamelCase ,				__UpperCamelCase							)
					    if self.use_ada_layer_norm
					    else nn.LayerNorm(__UpperCamelCase ,				elementwise_affine=__UpperCamelCase							)
					)
					snake_case_      		= Attention(
					    query_dim=__UpperCamelCase ,				cross_attention_dim=cross_attention_dim if not double_self_attention else None ,				heads=__UpperCamelCase ,				dim_head=__UpperCamelCase ,				dropout=__UpperCamelCase ,				bias=__UpperCamelCase ,				upcast_attention=__UpperCamelCase ,				)  # is self-attn if encoder_hidden_states is none
				else:
					snake_case_      		= None
					snake_case_      		= None

				# 3. Feed-forward
				snake_case_      		= nn.LayerNorm(__UpperCamelCase ,				elementwise_affine=__UpperCamelCase							)
				snake_case_      		= FeedForward(__UpperCamelCase ,				dropout=__UpperCamelCase ,				activation_fn=__UpperCamelCase ,				final_dropout=__UpperCamelCase							)

				# let chunk size default to None
				snake_case_      		= None
				snake_case_      		= 0


			def 			__lowerCAmelCase (    self ,				__UpperCamelCase ,				__UpperCamelCase							):

				"""simple docstring"""


				snake_case_      		= chunk_size
				snake_case_      		= dim







			def 			__lowerCAmelCase (    self ,				__UpperCamelCase ,				__UpperCamelCase = None ,				__UpperCamelCase = None ,				__UpperCamelCase = None ,				__UpperCamelCase = None ,				__UpperCamelCase = None ,				__UpperCamelCase = None ,				):

				"""simple docstring"""


				if self.use_ada_layer_norm:
					snake_case_      		= self.norma(__UpperCamelCase ,				__UpperCamelCase							)
				elif self.use_ada_layer_norm_zero:
					snake_case_		,       snake_case_		,       snake_case_		,       snake_case_		,       snake_case_      		= self.norma(
					    __UpperCamelCase ,				__UpperCamelCase ,				__UpperCamelCase ,				hidden_dtype=hidden_states.dtype							)
				else:
					snake_case_      		= self.norma(__UpperCamelCase							)

				snake_case_      		= cross_attention_kwargs if cross_attention_kwargs is not None else {}

				snake_case_      		= self.attna(
				    __UpperCamelCase ,				encoder_hidden_states=encoder_hidden_states if self.only_cross_attention else None ,				attention_mask=__UpperCamelCase ,				**__UpperCamelCase ,				)
				if self.use_ada_layer_norm_zero:
					snake_case_      		= gate_msa.unsqueeze(1							) * attn_output
				snake_case_      		= attn_output + hidden_states

				# 2. Cross-Attention
				if self.attna is not None:
					snake_case_      		= (
					    self.norma(__UpperCamelCase ,				__UpperCamelCase							) if self.use_ada_layer_norm else self.norma(__UpperCamelCase							)
					)

					snake_case_      		= self.attna(
					    __UpperCamelCase ,				encoder_hidden_states=__UpperCamelCase ,				attention_mask=__UpperCamelCase ,				**__UpperCamelCase ,				)
					snake_case_      		= attn_output + hidden_states

				# 3. Feed-forward
				snake_case_      		= self.norma(__UpperCamelCase							)

				if self.use_ada_layer_norm_zero:
					snake_case_      		= norm_hidden_states * (1 + scale_mlp[:, None]) + shift_mlp[:, None]

				if self._chunk_size is not None:
					# "feed_forward_chunk_size" can be used to save memory
					if norm_hidden_states.shape[self._chunk_dim] % self._chunk_size != 0:
						raise ValueError(
						    f"""`hidden_states` dimension to be chunked: {norm_hidden_states.shape[self._chunk_dim]} has to be divisible by chunk size: {self._chunk_size}. Make sure to set an appropriate `chunk_size` when calling `unet.enable_forward_chunking`."""							)

					snake_case_      		= norm_hidden_states.shape[self._chunk_dim] // self._chunk_size
					snake_case_      		= torch.cat(
					    [self.ff(__UpperCamelCase							) for hid_slice in norm_hidden_states.chunk(__UpperCamelCase ,				dim=self._chunk_dim							)] ,				dim=self._chunk_dim ,				)
				else:
					snake_case_      		= self.ff(__UpperCamelCase							)

				if self.use_ada_layer_norm_zero:
					snake_case_      		= gate_mlp.unsqueeze(1							) * ff_output

				snake_case_      		= ff_output + hidden_states

				return hidden_states







class SCREAMING_SNAKE_CASE  (		nn.Module							):






			"""simple docstring"""




			def __init__(    self ,				__UpperCamelCase ,				__UpperCamelCase = None ,				__UpperCamelCase = 4 ,				__UpperCamelCase = 0.0 ,				__UpperCamelCase = "geglu" ,				__UpperCamelCase = False ,				):

				"""simple docstring"""


				super().__init__()
				snake_case_      		= int(dim * mult							)
				snake_case_      		= dim_out if dim_out is not None else dim

				if activation_fn == "gelu":
					snake_case_      		= GELU(__UpperCamelCase ,				__UpperCamelCase							)
				if activation_fn == "gelu-approximate":
					snake_case_      		= GELU(__UpperCamelCase ,				__UpperCamelCase ,				approximate='tanh'							)
				elif activation_fn == "geglu":
					snake_case_      		= GEGLU(__UpperCamelCase ,				__UpperCamelCase							)
				elif activation_fn == "geglu-approximate":
					snake_case_      		= ApproximateGELU(__UpperCamelCase ,				__UpperCamelCase							)

				snake_case_      		= nn.ModuleList([]							)
				# project in
				self.net.append(__UpperCamelCase							)
				# project dropout
				self.net.append(nn.Dropout(__UpperCamelCase							)							)
				# project out
				self.net.append(nn.Linear(__UpperCamelCase ,				__UpperCamelCase							)							)
				# FF as used in Vision Transformer, MLP-Mixer, etc. have a final dropout
				if final_dropout:
					self.net.append(nn.Dropout(__UpperCamelCase							)							)







			def 			__lowerCAmelCase (    self ,				__UpperCamelCase							):

				"""simple docstring"""


				for module in self.net:
					snake_case_      		= module(__UpperCamelCase							)
				return hidden_states







class SCREAMING_SNAKE_CASE  (		nn.Module							):






			"""simple docstring"""




			def __init__(    self ,				__UpperCamelCase ,				__UpperCamelCase ,				__UpperCamelCase = "none"							):

				"""simple docstring"""


				super().__init__()
				snake_case_      		= nn.Linear(__UpperCamelCase ,				__UpperCamelCase							)
				snake_case_      		= approximate


			def 			__lowerCAmelCase (    self ,				__UpperCamelCase							):

				"""simple docstring"""


				if gate.device.type != "mps":
					return F.gelu(__UpperCamelCase ,				approximate=self.approximate							)
				# mps: gelu is not implemented for float16
				return F.gelu(gate.to(dtype=torch.floataa							) ,				approximate=self.approximate							).to(dtype=gate.dtype							)







			def 			__lowerCAmelCase (    self ,				__UpperCamelCase							):

				"""simple docstring"""


				snake_case_      		= self.proj(__UpperCamelCase							)
				snake_case_      		= self.gelu(__UpperCamelCase							)
				return hidden_states







class SCREAMING_SNAKE_CASE  (		nn.Module							):






			"""simple docstring"""




			def __init__(    self ,				__UpperCamelCase ,				__UpperCamelCase							):

				"""simple docstring"""


				super().__init__()
				snake_case_      		= nn.Linear(__UpperCamelCase ,				dim_out * 2							)


			def 			__lowerCAmelCase (    self ,				__UpperCamelCase							):

				"""simple docstring"""


				if gate.device.type != "mps":
					return F.gelu(__UpperCamelCase							)
				# mps: gelu is not implemented for float16
				return F.gelu(gate.to(dtype=torch.floataa							)							).to(dtype=gate.dtype							)







			def 			__lowerCAmelCase (    self ,				__UpperCamelCase							):

				"""simple docstring"""


				snake_case_		,       snake_case_      		= self.proj(__UpperCamelCase							).chunk(2 ,				dim=-1							)
				return hidden_states * self.gelu(__UpperCamelCase							)







class SCREAMING_SNAKE_CASE  (		nn.Module							):






			"""simple docstring"""




			def __init__(    self ,				__UpperCamelCase ,				__UpperCamelCase							):

				"""simple docstring"""


				super().__init__()
				snake_case_      		= nn.Linear(__UpperCamelCase ,				__UpperCamelCase							)







			def 			__lowerCAmelCase (    self ,				__UpperCamelCase							):

				"""simple docstring"""


				snake_case_      		= self.proj(__UpperCamelCase							)
				return x * torch.sigmoid(1.702 * x							)







class SCREAMING_SNAKE_CASE  (		nn.Module							):






			"""simple docstring"""




			def __init__(    self ,				__UpperCamelCase ,				__UpperCamelCase							):

				"""simple docstring"""


				super().__init__()
				snake_case_      		= nn.Embedding(__UpperCamelCase ,				__UpperCamelCase							)
				snake_case_      		= nn.SiLU()
				snake_case_      		= nn.Linear(__UpperCamelCase ,				embedding_dim * 2							)
				snake_case_      		= nn.LayerNorm(__UpperCamelCase ,				elementwise_affine=__UpperCamelCase							)







			def 			__lowerCAmelCase (    self ,				__UpperCamelCase ,				__UpperCamelCase							):

				"""simple docstring"""


				snake_case_      		= self.linear(self.silu(self.emb(__UpperCamelCase							)							)							)
				snake_case_		,       snake_case_      		= torch.chunk(__UpperCamelCase ,				2							)
				snake_case_      		= self.norm(__UpperCamelCase							) * (1 + scale) + shift
				return x







class SCREAMING_SNAKE_CASE  (		nn.Module							):






			"""simple docstring"""




			def __init__(    self ,				__UpperCamelCase ,				__UpperCamelCase							):

				"""simple docstring"""


				super().__init__()

				snake_case_      		= CombinedTimestepLabelEmbeddings(__UpperCamelCase ,				__UpperCamelCase							)

				snake_case_      		= nn.SiLU()
				snake_case_      		= nn.Linear(__UpperCamelCase ,				6 * embedding_dim ,				bias=__UpperCamelCase							)
				snake_case_      		= nn.LayerNorm(__UpperCamelCase ,				elementwise_affine=__UpperCamelCase ,				eps=1E-6							)







			def 			__lowerCAmelCase (    self ,				__UpperCamelCase ,				__UpperCamelCase ,				__UpperCamelCase ,				__UpperCamelCase=None							):

				"""simple docstring"""


				snake_case_      		= self.linear(self.silu(self.emb(__UpperCamelCase ,				__UpperCamelCase ,				hidden_dtype=__UpperCamelCase							)							)							)
				snake_case_		,       snake_case_		,       snake_case_		,       snake_case_		,       snake_case_		,       snake_case_      		= emb.chunk(6 ,				dim=1							)
				snake_case_      		= self.norm(__UpperCamelCase							) * (1 + scale_msa[:, None]) + shift_msa[:, None]
				return x, gate_msa, shift_mlp, scale_mlp, gate_mlp







class SCREAMING_SNAKE_CASE  (		nn.Module							):






			"""simple docstring"""




			def __init__(    self ,				__UpperCamelCase ,				__UpperCamelCase ,				__UpperCamelCase ,				__UpperCamelCase = None ,				__UpperCamelCase = 1E-5							):

				"""simple docstring"""


				super().__init__()
				snake_case_      		= num_groups
				snake_case_      		= eps

				if act_fn is None:
					snake_case_      		= None
				else:
					snake_case_      		= get_activation(__UpperCamelCase							)

				snake_case_      		= nn.Linear(__UpperCamelCase ,				out_dim * 2							)







			def 			__lowerCAmelCase (    self ,				__UpperCamelCase ,				__UpperCamelCase							):

				"""simple docstring"""


				if self.act:
					snake_case_      		= self.act(__UpperCamelCase							)
				snake_case_      		= self.linear(__UpperCamelCase							)
				snake_case_      		= emb[:, :, None, None]
				snake_case_		,       snake_case_      		= emb.chunk(2 ,				dim=1							)

				snake_case_      		= F.group_norm(__UpperCamelCase ,				self.num_groups ,				eps=self.eps							)
				snake_case_      		= x * (1 + scale) + shift
				return x



import unittest

import numpy as np
import torch

from diffusers import ScoreSdeVePipeline, ScoreSdeVeScheduler, UNetaDModel
from diffusers.utils.testing_utils import enable_full_determinism, require_torch, slow, torch_device


enable_full_determinism()


class SCREAMING_SNAKE_CASE  (		unittest.TestCase							):






			"""simple docstring"""




			@property
			def 			__lowerCAmelCase (    self							):

				"""simple docstring"""


				torch.manual_seed(0							)
				snake_case_      		= UNetaDModel(
				    block_out_channels=(32, 64) ,				layers_per_block=2 ,				sample_size=32 ,				in_channels=3 ,				out_channels=3 ,				down_block_types=('DownBlock2D', 'AttnDownBlock2D') ,				up_block_types=('AttnUpBlock2D', 'UpBlock2D') ,				)
				return model







			def 			__lowerCAmelCase (    self							):

				"""simple docstring"""


				snake_case_      		= self.dummy_uncond_unet
				snake_case_      		= ScoreSdeVeScheduler()

				snake_case_      		= ScoreSdeVePipeline(unet=__UpperCamelCase ,				scheduler=__UpperCamelCase							)
				sde_ve.to(__UpperCamelCase							)
				sde_ve.set_progress_bar_config(disable=__UpperCamelCase							)

				snake_case_      		= torch.manual_seed(0							)
				snake_case_      		= sde_ve(num_inference_steps=2 ,				output_type='numpy' ,				generator=__UpperCamelCase							).images

				snake_case_      		= torch.manual_seed(0							)
				snake_case_      		= sde_ve(num_inference_steps=2 ,				output_type='numpy' ,				generator=__UpperCamelCase ,				return_dict=__UpperCamelCase							)[
				    0
				]

				snake_case_      		= image[0, -3:, -3:, -1]
				snake_case_      		= image_from_tuple[0, -3:, -3:, -1]

				assert image.shape == (1, 32, 32, 3)
				snake_case_      		= np.array([0.0, 1.0, 0.0, 0.0, 0.0, 1.0, 0.0, 0.0, 0.0]							)

				assert np.abs(image_slice.flatten() - expected_slice							).max() < 1E-2
				assert np.abs(image_from_tuple_slice.flatten() - expected_slice							).max() < 1E-2







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






			"""simple docstring"""




			def 			__lowerCAmelCase (    self							):

				"""simple docstring"""


				snake_case_      		= 'google/ncsnpp-church-256'
				snake_case_      		= UNetaDModel.from_pretrained(__UpperCamelCase							)

				snake_case_      		= ScoreSdeVeScheduler.from_pretrained(__UpperCamelCase							)

				snake_case_      		= ScoreSdeVePipeline(unet=__UpperCamelCase ,				scheduler=__UpperCamelCase							)
				sde_ve.to(__UpperCamelCase							)
				sde_ve.set_progress_bar_config(disable=__UpperCamelCase							)

				snake_case_      		= torch.manual_seed(0							)
				snake_case_      		= sde_ve(num_inference_steps=10 ,				output_type='numpy' ,				generator=__UpperCamelCase							).images

				snake_case_      		= image[0, -3:, -3:, -1]

				assert image.shape == (1, 2_56, 2_56, 3)

				snake_case_      		= np.array([0.0, 1.0, 1.0, 1.0, 1.0, 1.0, 1.0, 0.0, 0.0]							)

				assert np.abs(image_slice.flatten() - expected_slice							).max() < 1E-2
