



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 	__snake_case   ( nn.Module ):
		def __init__(							self    ,a_    ,a_    ,a_    ,a_=0.0    ,a_ = None    ,a_ = "geglu"    ,a_ = None    ,a_ = False    ,a_ = False    ,a_ = False    ,a_ = False    ,a_ = True    ,a_ = "layer_norm"    ,a_ = False    ,):







						"""simple docstring"""
						super().__init__()
						lowerCAmelCase__		  =  only_cross_attention

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

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

						# let chunk size default to None
						lowerCAmelCase__		  =  None
						lowerCAmelCase__		  =  0
		def 						SCREAMING_SNAKE_CASE_					(							self    ,a_    ,a_						):







						"""simple docstring"""
						# Sets chunk feed-forward
						lowerCAmelCase__		  =  chunk_size
						lowerCAmelCase__		  =  dim



		def 						SCREAMING_SNAKE_CASE_					(							self    ,a_    ,a_ = None    ,a_ = None    ,a_ = None    ,a_ = None    ,a_ = None    ,a_ = None    ,):







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

						lowerCAmelCase__		  =  cross_attention_kwargs if cross_attention_kwargs is not None else {}

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

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

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

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

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

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

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

						lowerCAmelCase__		  =  ff_output + hidden_states

						return hidden_states



class 	__snake_case   ( nn.Module ):
		def __init__(							self    ,a_    ,a_ = None    ,a_ = 4    ,a_ = 0.0    ,a_ = "geglu"    ,a_ = False    ,):







						"""simple docstring"""
						super().__init__()
						lowerCAmelCase__		  =  int(dim * mult						)
						lowerCAmelCase__		  =  dim_out if dim_out is not None else dim

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

						lowerCAmelCase__		  =  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 						SCREAMING_SNAKE_CASE_					(							self    ,a_						):







						"""simple docstring"""
						for module in self.net:
										lowerCAmelCase__		  =  module(a_						)
						return hidden_states



class 	__snake_case   ( nn.Module ):
		def __init__(							self    ,a_    ,a_    ,a_ = "none"						):







						"""simple docstring"""
						super().__init__()
						lowerCAmelCase__		  =  nn.Linear(a_    ,a_						)
						lowerCAmelCase__		  =  approximate
		def 						SCREAMING_SNAKE_CASE_					(							self    ,a_						):







						"""simple docstring"""
						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 						SCREAMING_SNAKE_CASE_					(							self    ,a_						):







						"""simple docstring"""
						lowerCAmelCase__		  =  self.proj(a_						)
						lowerCAmelCase__		  =  self.gelu(a_						)
						return hidden_states



class 	__snake_case   ( nn.Module ):
		def __init__(							self    ,a_    ,a_						):







						"""simple docstring"""
						super().__init__()
						lowerCAmelCase__		  =  nn.Linear(a_    ,dim_out * 2						)
		def 						SCREAMING_SNAKE_CASE_					(							self    ,a_						):







						"""simple docstring"""
						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 						SCREAMING_SNAKE_CASE_					(							self    ,a_						):







						"""simple docstring"""
						lowerCAmelCase__		  =  self.proj(a_						).chunk(2    ,dim=-1						)
						return hidden_states * self.gelu(a_						)



class 	__snake_case   ( nn.Module ):
		def __init__(							self    ,a_    ,a_						):







						"""simple docstring"""
						super().__init__()
						lowerCAmelCase__		  =  nn.Linear(a_    ,a_						)



		def 						SCREAMING_SNAKE_CASE_					(							self    ,a_						):







						"""simple docstring"""
						lowerCAmelCase__		  =  self.proj(a_						)
						return x * torch.sigmoid(1.702 * x						)



class 	__snake_case   ( nn.Module ):
		def __init__(							self    ,a_    ,a_						):







						"""simple docstring"""
						super().__init__()
						lowerCAmelCase__		  =  nn.Embedding(a_    ,a_						)
						lowerCAmelCase__		  =  nn.SiLU()
						lowerCAmelCase__		  =  nn.Linear(a_    ,embedding_dim * 2						)
						lowerCAmelCase__		  =  nn.LayerNorm(a_    ,elementwise_affine=a_						)



		def 						SCREAMING_SNAKE_CASE_					(							self    ,a_    ,a_						):







						"""simple docstring"""
						lowerCAmelCase__		  =  self.linear(self.silu(self.emb(a_						)						)						)
						lowerCAmelCase__		  =  torch.chunk(a_    ,2						)
						lowerCAmelCase__		  =  self.norm(a_						) * (1 + scale) + shift
						return x



class 	__snake_case   ( nn.Module ):
		def __init__(							self    ,a_    ,a_						):







						"""simple docstring"""
						super().__init__()

						lowerCAmelCase__		  =  CombinedTimestepLabelEmbeddings(a_    ,a_						)

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



		def 						SCREAMING_SNAKE_CASE_					(							self    ,a_    ,a_    ,a_    ,a_=None						):







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



class 	__snake_case   ( nn.Module ):
		def __init__(							self    ,a_    ,a_    ,a_    ,a_ = None    ,a_ = 1e-5						):







						"""simple docstring"""
						super().__init__()
						lowerCAmelCase__		  =  num_groups
						lowerCAmelCase__		  =  eps

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

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



		def 						SCREAMING_SNAKE_CASE_					(							self    ,a_    ,a_						):







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

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




"""simple docstring"""
# limitations under the License.

# NOTE: This file is deprecated and will be removed in a future version.
# It only exists so that temporarely `from diffusers.pipelines import DiffusionPipeline` works

from .pipelines import DiffusionPipeline, ImagePipelineOutput  # noqa: F401
from .utils import deprecate


deprecate(
    """pipelines_utils""",
    """0.22.0""",
    """Importing `DiffusionPipeline` or `ImagePipelineOutput` from diffusers.pipeline_utils is deprecated. Please import from diffusers.pipelines.pipeline_utils instead.""",
    standard_warn=False,
    stacklevel=3,
)

