
'''simple docstring'''
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   a_ (      nn.Module					):




			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					,   ):
				super().__init__()
				a_      				=						only_cross_attention

				a_      				=						(num_embeds_ada_norm is not None) and norm_type == """ada_norm_zero"""
				a_      				=						(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:
					a_      				=						AdaLayerNorm(UpperCAmelCase					,   UpperCAmelCase	)
				elif self.use_ada_layer_norm_zero:
					a_      				=						AdaLayerNormZero(UpperCAmelCase					,   UpperCAmelCase	)
				else:
					a_      				=						nn.LayerNorm(UpperCAmelCase					,   elementwise_affine=UpperCAmelCase	)
				a_      				=						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.
					a_      				=						(
					    AdaLayerNorm(UpperCAmelCase					,   UpperCAmelCase	)
					    if self.use_ada_layer_norm
					    else nn.LayerNorm(UpperCAmelCase					,   elementwise_affine=UpperCAmelCase	)
					)
					a_      				=						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:
					a_      				=						None
					a_      				=						None

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

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




			def 					lowerCAmelCase__					(       self					,   UpperCAmelCase					,   UpperCAmelCase	):
				# Sets chunk feed-forward
				a_      				=						chunk_size
				a_      				=						dim






			def 					lowerCAmelCase__					(       self					,   UpperCAmelCase					,   UpperCAmelCase = None					,   UpperCAmelCase = None					,   UpperCAmelCase = None					,   UpperCAmelCase = None					,   UpperCAmelCase = None					,   UpperCAmelCase = None					,   ):
				# Notice that normalization is always applied before the real computation in the following blocks.
				# 1. Self-Attention
				if self.use_ada_layer_norm:
					a_      				=						self.norma(UpperCAmelCase					,   UpperCAmelCase	)
				elif self.use_ada_layer_norm_zero:
					a_				,  a_				,  a_				,  a_				,  a_      				=						self.norma(
					    UpperCAmelCase					,   UpperCAmelCase					,   UpperCAmelCase					,   hidden_dtype=hidden_states.dtype	)
				else:
					a_      				=						self.norma(UpperCAmelCase	)

				a_      				=						cross_attention_kwargs if cross_attention_kwargs is not None else {}

				a_      				=						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:
					a_      				=						gate_msa.unsqueeze(1	) * attn_output
				a_      				=						attn_output + hidden_states

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

					a_      				=						self.attna(
					    UpperCAmelCase					,   encoder_hidden_states=UpperCAmelCase					,   attention_mask=UpperCAmelCase					,   **UpperCAmelCase					,   )
					a_      				=						attn_output + hidden_states

				# 3. Feed-forward
				a_      				=						self.norma(UpperCAmelCase	)

				if self.use_ada_layer_norm_zero:
					a_      				=						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`.'''	)

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

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

				a_      				=						ff_output + hidden_states

				return hidden_states







class   a_ (      nn.Module					):




			def __init__(       self					,   UpperCAmelCase					,   UpperCAmelCase = None					,   UpperCAmelCase = 4					,   UpperCAmelCase = 0.0					,   UpperCAmelCase = "geglu"					,   UpperCAmelCase = False					,   ):
				super().__init__()
				a_      				=						int(dim * mult	)
				a_      				=						dim_out if dim_out is not None else dim

				if activation_fn == "gelu":
					a_      				=						GELU(UpperCAmelCase					,   UpperCAmelCase	)
				if activation_fn == "gelu-approximate":
					a_      				=						GELU(UpperCAmelCase					,   UpperCAmelCase					,   approximate="""tanh"""	)
				elif activation_fn == "geglu":
					a_      				=						GEGLU(UpperCAmelCase					,   UpperCAmelCase	)
				elif activation_fn == "geglu-approximate":
					a_      				=						ApproximateGELU(UpperCAmelCase					,   UpperCAmelCase	)

				a_      				=						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	):
				for module in self.net:
					a_      				=						module(UpperCAmelCase	)
				return hidden_states







class   a_ (      nn.Module					):




			def __init__(       self					,   UpperCAmelCase					,   UpperCAmelCase					,   UpperCAmelCase = "none"	):
				super().__init__()
				a_      				=						nn.Linear(UpperCAmelCase					,   UpperCAmelCase	)
				a_      				=						approximate




			def 					lowerCAmelCase__					(       self					,   UpperCAmelCase	):
				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	):
				a_      				=						self.proj(UpperCAmelCase	)
				a_      				=						self.gelu(UpperCAmelCase	)
				return hidden_states







class   a_ (      nn.Module					):




			def __init__(       self					,   UpperCAmelCase					,   UpperCAmelCase	):
				super().__init__()
				a_      				=						nn.Linear(UpperCAmelCase					,   dim_out * 2	)




			def 					lowerCAmelCase__					(       self					,   UpperCAmelCase	):
				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	):
				a_				,  a_      				=						self.proj(UpperCAmelCase	).chunk(2					,   dim=-1	)
				return hidden_states * self.gelu(UpperCAmelCase	)







class   a_ (      nn.Module					):




			def __init__(       self					,   UpperCAmelCase					,   UpperCAmelCase	):
				super().__init__()
				a_      				=						nn.Linear(UpperCAmelCase					,   UpperCAmelCase	)






			def 					lowerCAmelCase__					(       self					,   UpperCAmelCase	):
				a_      				=						self.proj(UpperCAmelCase	)
				return x * torch.sigmoid(1.7_02 * x	)







class   a_ (      nn.Module					):




			def __init__(       self					,   UpperCAmelCase					,   UpperCAmelCase	):
				super().__init__()
				a_      				=						nn.Embedding(UpperCAmelCase					,   UpperCAmelCase	)
				a_      				=						nn.SiLU()
				a_      				=						nn.Linear(UpperCAmelCase					,   embedding_dim * 2	)
				a_      				=						nn.LayerNorm(UpperCAmelCase					,   elementwise_affine=UpperCAmelCase	)






			def 					lowerCAmelCase__					(       self					,   UpperCAmelCase					,   UpperCAmelCase	):
				a_      				=						self.linear(self.silu(self.emb(UpperCAmelCase	)	)	)
				a_				,  a_      				=						torch.chunk(UpperCAmelCase					,   2	)
				a_      				=						self.norm(UpperCAmelCase	) * (1 + scale) + shift
				return x







class   a_ (      nn.Module					):




			def __init__(       self					,   UpperCAmelCase					,   UpperCAmelCase	):
				super().__init__()

				a_      				=						CombinedTimestepLabelEmbeddings(UpperCAmelCase					,   UpperCAmelCase	)

				a_      				=						nn.SiLU()
				a_      				=						nn.Linear(UpperCAmelCase					,   6 * embedding_dim					,   bias=UpperCAmelCase	)
				a_      				=						nn.LayerNorm(UpperCAmelCase					,   elementwise_affine=UpperCAmelCase					,   eps=1e-6	)






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







class   a_ (      nn.Module					):




			def __init__(       self					,   UpperCAmelCase					,   UpperCAmelCase					,   UpperCAmelCase					,   UpperCAmelCase = None					,   UpperCAmelCase = 1e-5	):
				super().__init__()
				a_      				=						num_groups
				a_      				=						eps

				if act_fn is None:
					a_      				=						None
				else:
					a_      				=						get_activation(UpperCAmelCase	)

				a_      				=						nn.Linear(UpperCAmelCase					,   out_dim * 2	)






			def 					lowerCAmelCase__					(       self					,   UpperCAmelCase					,   UpperCAmelCase	):
				if self.act:
					a_      				=						self.act(UpperCAmelCase	)
				a_      				=						self.linear(UpperCAmelCase	)
				a_      				=						emb[:, :, None, None]
				a_				,  a_      				=						emb.chunk(2					,   dim=1	)

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


'''simple docstring'''
import unittest

import torch

from diffusers import DDIMScheduler, DDPMScheduler, UNetaDModel
from diffusers.training_utils import set_seed
from diffusers.utils.testing_utils import slow


lowercase__											=False







class   a_ (      unittest.TestCase					):




			def 					lowerCAmelCase__					(       self					,   UpperCAmelCase=32	):
				set_seed(0	)
				a_      				=						UNetaDModel(sample_size=UpperCAmelCase					,   in_channels=3					,   out_channels=3	)
				a_      				=						torch.optim.SGD(model.parameters()					,   lr=0.00_01	)
				return model, optimizer






			@slow
			def 					lowerCAmelCase__					(       self	):
				a_      				=						"""cpu"""  # ensure full determinism without setting the CUBLAS_WORKSPACE_CONFIG env variable
				a_      				=						DDPMScheduler(
				    num_train_timesteps=10_00					,   beta_start=0.00_01					,   beta_end=0.02					,   beta_schedule="""linear"""					,   clip_sample=UpperCAmelCase					,   )
				a_      				=						DDIMScheduler(
				    num_train_timesteps=10_00					,   beta_start=0.00_01					,   beta_end=0.02					,   beta_schedule="""linear"""					,   clip_sample=UpperCAmelCase					,   )

				assert ddpm_scheduler.config.num_train_timesteps == ddim_scheduler.config.num_train_timesteps

				# shared batches for DDPM and DDIM
				set_seed(0	)
				a_      				=						[torch.randn((4, 3, 32, 32)	).clip(-1					,   1	).to(UpperCAmelCase	) for _ in range(4	)]
				a_      				=						[torch.randn((4, 3, 32, 32)	).to(UpperCAmelCase	) for _ in range(4	)]
				a_      				=						[torch.randint(0					,   10_00					,   (4,)	).long().to(UpperCAmelCase	) for _ in range(4	)]

				# train with a DDPM scheduler
				a_				,  a_      				=						self.get_model_optimizer(resolution=32	)
				model.train().to(UpperCAmelCase	)
				for i in range(4	):
					optimizer.zero_grad()
					a_      				=						ddpm_scheduler.add_noise(clean_images[i]					,   noise[i]					,   timesteps[i]	)
					a_      				=						model(UpperCAmelCase					,   timesteps[i]	).sample
					a_      				=						torch.nn.functional.mse_loss(UpperCAmelCase					,   noise[i]	)
					loss.backward()
					optimizer.step()
				del model, optimizer

				# recreate the model and optimizer, and retry with DDIM
				a_				,  a_      				=						self.get_model_optimizer(resolution=32	)
				model.train().to(UpperCAmelCase	)
				for i in range(4	):
					optimizer.zero_grad()
					a_      				=						ddim_scheduler.add_noise(clean_images[i]					,   noise[i]					,   timesteps[i]	)
					a_      				=						model(UpperCAmelCase					,   timesteps[i]	).sample
					a_      				=						torch.nn.functional.mse_loss(UpperCAmelCase					,   noise[i]	)
					loss.backward()
					optimizer.step()
				del model, optimizer

				self.assertTrue(torch.allclose(UpperCAmelCase					,   UpperCAmelCase					,   atol=1e-5	)	)
				self.assertTrue(torch.allclose(UpperCAmelCase					,   UpperCAmelCase					,   atol=1e-5	)	)
