





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 			UpperCamelCase_		(  nn.Module					):




					def __init__(		self				,						lowerCAmelCase_				,						lowerCAmelCase_				,						lowerCAmelCase_				,						lowerCAmelCase_=0.0				,						lowerCAmelCase_ = None				,						lowerCAmelCase_ = "geglu"				,						lowerCAmelCase_ = None				,						lowerCAmelCase_ = False				,						lowerCAmelCase_ = False				,						lowerCAmelCase_ = False				,						lowerCAmelCase_ = False				,						lowerCAmelCase_ = True				,						lowerCAmelCase_ = "layer_norm"				,						lowerCAmelCase_ = False				,						)		->							List[str]:
										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(__a				,						__a  )
										elif self.use_ada_layer_norm_zero:
															_snake_case									=				AdaLayerNormZero(__a				,						__a  )
										else:
															_snake_case									=				nn.LayerNorm(__a				,						elementwise_affine=__a  )
										_snake_case									=				Attention(
										    query_dim=__a				,						heads=__a				,						dim_head=__a				,						dropout=__a				,						bias=__a				,						cross_attention_dim=cross_attention_dim if only_cross_attention else None				,						upcast_attention=__a				,						)

										# 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(__a				,						__a  )
															    if self.use_ada_layer_norm
															    else nn.LayerNorm(__a				,						elementwise_affine=__a  )
															)
															_snake_case									=				Attention(
															    query_dim=__a				,						cross_attention_dim=cross_attention_dim if not double_self_attention else None				,						heads=__a				,						dim_head=__a				,						dropout=__a				,						bias=__a				,						upcast_attention=__a				,						)  # is self-attn if encoder_hidden_states is none
										else:
															_snake_case									=				None
															_snake_case									=				None

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

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




					def 	lowerCAmelCase			(		self				,						lowerCAmelCase_				,						lowerCAmelCase_  )		->							Tuple:
										# Sets chunk feed-forward
										_snake_case									=				chunk_size
										_snake_case									=				dim






					def 	lowerCAmelCase			(		self				,						lowerCAmelCase_				,						lowerCAmelCase_ = None				,						lowerCAmelCase_ = None				,						lowerCAmelCase_ = None				,						lowerCAmelCase_ = None				,						lowerCAmelCase_ = None				,						lowerCAmelCase_ = None				,						)		->							Optional[Any]:
										# Notice that normalization is always applied before the real computation in the following blocks.
										# 1. Self-Attention
										if self.use_ada_layer_norm:
															_snake_case									=				self.norma(__a				,						__a  )
										elif self.use_ada_layer_norm_zero:
															_snake_case									=				self.norma(
															    __a				,						__a				,						__a				,						hidden_dtype=hidden_states.dtype  )
										else:
															_snake_case									=				self.norma(__a  )

										_snake_case									=				cross_attention_kwargs if cross_attention_kwargs is not None else {}

										_snake_case									=				self.attna(
										    __a				,						encoder_hidden_states=encoder_hidden_states if self.only_cross_attention else None				,						attention_mask=__a				,						**__a				,						)
										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(__a				,						__a  ) if self.use_ada_layer_norm else self.norma(__a  )
															)

															_snake_case									=				self.attna(
															    __a				,						encoder_hidden_states=__a				,						attention_mask=__a				,						**__a				,						)
															_snake_case									=				attn_output + hidden_states

										# 3. Feed-forward
										_snake_case									=				self.norma(__a  )

										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(__a  ) for hid_slice in norm_hidden_states.chunk(__a				,						dim=self._chunk_dim  )]				,						dim=self._chunk_dim				,						)
										else:
															_snake_case									=				self.ff(__a  )

										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 			UpperCamelCase_		(  nn.Module					):




					def __init__(		self				,						lowerCAmelCase_				,						lowerCAmelCase_ = None				,						lowerCAmelCase_ = 4				,						lowerCAmelCase_ = 0.0				,						lowerCAmelCase_ = "geglu"				,						lowerCAmelCase_ = False				,						)		->							str:
										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(__a				,						__a  )
										if activation_fn == "gelu-approximate":
															_snake_case									=				GELU(__a				,						__a				,						approximate='tanh'  )
										elif activation_fn == "geglu":
															_snake_case									=				GEGLU(__a				,						__a  )
										elif activation_fn == "geglu-approximate":
															_snake_case									=				ApproximateGELU(__a				,						__a  )

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






					def 	lowerCAmelCase			(		self				,						lowerCAmelCase_  )		->							List[str]:
										for module in self.net:
															_snake_case									=				module(__a  )
										return hidden_states





class 			UpperCamelCase_		(  nn.Module					):




					def __init__(		self				,						lowerCAmelCase_				,						lowerCAmelCase_				,						lowerCAmelCase_ = "none"  )		->							List[str]:
										super().__init__()
										_snake_case									=				nn.Linear(__a				,						__a  )
										_snake_case									=				approximate




					def 	lowerCAmelCase			(		self				,						lowerCAmelCase_  )		->							Dict:
										if gate.device.type != "mps":
															return F.gelu(__a				,						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				,						lowerCAmelCase_  )		->							Tuple:
										_snake_case									=				self.proj(__a  )
										_snake_case									=				self.gelu(__a  )
										return hidden_states





class 			UpperCamelCase_		(  nn.Module					):




					def __init__(		self				,						lowerCAmelCase_				,						lowerCAmelCase_  )		->							str:
										super().__init__()
										_snake_case									=				nn.Linear(__a				,						dim_out * 2  )




					def 	lowerCAmelCase			(		self				,						lowerCAmelCase_  )		->							List[str]:
										if gate.device.type != "mps":
															return F.gelu(__a  )
										# mps: gelu is not implemented for float16
										return F.gelu(gate.to(dtype=torch.floataa  )  ).to(dtype=gate.dtype  )






					def 	lowerCAmelCase			(		self				,						lowerCAmelCase_  )		->							Optional[int]:
										_snake_case									=				self.proj(__a  ).chunk(2				,						dim=-1  )
										return hidden_states * self.gelu(__a  )





class 			UpperCamelCase_		(  nn.Module					):




					def __init__(		self				,						lowerCAmelCase_				,						lowerCAmelCase_  )		->							Any:
										super().__init__()
										_snake_case									=				nn.Linear(__a				,						__a  )






					def 	lowerCAmelCase			(		self				,						lowerCAmelCase_  )		->							str:
										_snake_case									=				self.proj(__a  )
										return x * torch.sigmoid(1.7_02 * x  )





class 			UpperCamelCase_		(  nn.Module					):




					def __init__(		self				,						lowerCAmelCase_				,						lowerCAmelCase_  )		->							Union[str, Any]:
										super().__init__()
										_snake_case									=				nn.Embedding(__a				,						__a  )
										_snake_case									=				nn.SiLU()
										_snake_case									=				nn.Linear(__a				,						embedding_dim * 2  )
										_snake_case									=				nn.LayerNorm(__a				,						elementwise_affine=__a  )






					def 	lowerCAmelCase			(		self				,						lowerCAmelCase_				,						lowerCAmelCase_  )		->							List[Any]:
										_snake_case									=				self.linear(self.silu(self.emb(__a  )  )  )
										_snake_case									=				torch.chunk(__a				,						2  )
										_snake_case									=				self.norm(__a  ) * (1 + scale) + shift
										return x





class 			UpperCamelCase_		(  nn.Module					):




					def __init__(		self				,						lowerCAmelCase_				,						lowerCAmelCase_  )		->							List[Any]:
										super().__init__()

										_snake_case									=				CombinedTimestepLabelEmbeddings(__a				,						__a  )

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






					def 	lowerCAmelCase			(		self				,						lowerCAmelCase_				,						lowerCAmelCase_				,						lowerCAmelCase_				,						lowerCAmelCase_=None  )		->							int:
										_snake_case									=				self.linear(self.silu(self.emb(__a				,						__a				,						hidden_dtype=__a  )  )  )
										_snake_case									=				emb.chunk(6				,						dim=1  )
										_snake_case									=				self.norm(__a  ) * (1 + scale_msa[:, None]) + shift_msa[:, None]
										return x, gate_msa, shift_mlp, scale_mlp, gate_mlp





class 			UpperCamelCase_		(  nn.Module					):




					def __init__(		self				,						lowerCAmelCase_				,						lowerCAmelCase_				,						lowerCAmelCase_				,						lowerCAmelCase_ = None				,						lowerCAmelCase_ = 1E-5  )		->							Dict:
										super().__init__()
										_snake_case									=				num_groups
										_snake_case									=				eps

										if act_fn is None:
															_snake_case									=				None
										else:
															_snake_case									=				get_activation(__a  )

										_snake_case									=				nn.Linear(__a				,						out_dim * 2  )






					def 	lowerCAmelCase			(		self				,						lowerCAmelCase_				,						lowerCAmelCase_  )		->							Tuple:
										if self.act:
															_snake_case									=				self.act(__a  )
										_snake_case									=				self.linear(__a  )
										_snake_case									=				emb[:, :, None, None]
										_snake_case									=				emb.chunk(2				,						dim=1  )

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










import unittest

from accelerate import debug_launcher
from accelerate.test_utils import require_cpu, test_ops, test_script
@require_cpu
class 			UpperCamelCase_		(  unittest.TestCase					):




					def 	lowerCAmelCase			(		self  )		->							List[Any]:
										debug_launcher(test_script.main  )






					def 	lowerCAmelCase			(		self  )		->							Optional[Any]:
										debug_launcher(test_ops.main  )



