





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      ):





       '''simple docstring'''







       def __init__(  self			:   List[Any]							,    A			:   int							,    A			:   int							,    A			:   int							,    A			:   List[str]=0.0							,    A			:   Optional[int] = None							,    A			:   str = "geglu"							,    A			:   Optional[int] = None							,    A			:   bool = False							,    A			:   bool = False							,    A			:   bool = False							,    A			:   bool = False							,    A			:   bool = True							,    A			:   str = "layer_norm"							,    A			:   bool = False							,    ):
         super().__init__()
         __snake_case:						Tuple			  =						only_cross_attention

         __snake_case:						List[str]			  =						(num_embeds_ada_norm is not None) and norm_type == """ada_norm_zero"""
         __snake_case:						Dict			  =						(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:						Any			  =						AdaLayerNorm(A							,    A		)
         elif self.use_ada_layer_norm_zero:
           __snake_case:						int			  =						AdaLayerNormZero(A							,    A		)
         else:
           __snake_case:						str			  =						nn.LayerNorm(A							,    elementwise_affine=A		)
         __snake_case:						Any			  =						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:						Union[str, Any]			  =						(
               AdaLayerNorm(A							,    A		)
               if self.use_ada_layer_norm
               else nn.LayerNorm(A							,    elementwise_affine=A		)
           )
           __snake_case:						Tuple			  =						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:						Optional[int]			  =						None
           __snake_case:						Tuple			  =						None

         # 3. Feed-forward
         __snake_case:						Dict			  =						nn.LayerNorm(A							,    elementwise_affine=A		)
         __snake_case:						Tuple			  =						FeedForward(A							,    dropout=A							,    activation_fn=A							,    final_dropout=A		)

         # let chunk size default to None
         __snake_case:						int			  =						None
         __snake_case:						int			  =						0




       def 							UpperCAmelCase__			(  self			:   str							,    A			:   Optional[int]							,    A			:   int		):
         # Sets chunk feed-forward
         __snake_case:						Optional[Any]			  =						chunk_size
         __snake_case:						Union[str, Any]			  =						dim






       def 							UpperCAmelCase__			(  self			:   Tuple							,    A			:   torch.FloatTensor							,    A			:   Optional[torch.FloatTensor] = None							,    A			:   Optional[torch.FloatTensor] = None							,    A			:   Optional[torch.FloatTensor] = None							,    A			:   Optional[torch.LongTensor] = None							,    A			:   Dict[str, Any] = None							,    A			:   Optional[torch.LongTensor] = None							,    ):
         # 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:						str			  =						self.norma(A							,    A		)
         elif self.use_ada_layer_norm_zero:
           __snake_case  ,       __snake_case  ,       __snake_case  ,       __snake_case  ,       __snake_case:						Any			  =						self.norma(
               A							,    A							,    A							,    hidden_dtype=hidden_states.dtype		)
         else:
           __snake_case:						Dict			  =						self.norma(A		)

         __snake_case:						Optional[Any]			  =						cross_attention_kwargs if cross_attention_kwargs is not None else {}

         __snake_case:						List[str]			  =						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:						str			  =						gate_msa.unsqueeze(1		) * attn_output
         __snake_case:						Optional[Any]			  =						attn_output + hidden_states

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

           __snake_case:						Union[str, Any]			  =						self.attna(
               A							,    encoder_hidden_states=A							,    attention_mask=A							,    **A							,    )
           __snake_case:						Optional[Any]			  =						attn_output + hidden_states

         # 3. Feed-forward
         __snake_case:						Union[str, Any]			  =						self.norma(A		)

         if self.use_ada_layer_norm_zero:
           __snake_case:						List[str]			  =						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:						Any			  =						norm_hidden_states.shape[self._chunk_dim] // self._chunk_size
           __snake_case:						int			  =						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:						Optional[int]			  =						self.ff(A		)

         if self.use_ada_layer_norm_zero:
           __snake_case:						List[Any]			  =						gate_mlp.unsqueeze(1		) * ff_output

         __snake_case:						int			  =						ff_output + hidden_states

         return hidden_states







class    __snake_case	(							nn.Module      ):





       '''simple docstring'''







       def __init__(  self			:   List[str]							,    A			:   int							,    A			:   Optional[int] = None							,    A			:   int = 4							,    A			:   float = 0.0							,    A			:   str = "geglu"							,    A			:   bool = False							,    ):
         super().__init__()
         __snake_case:						Union[str, Any]			  =						int(dim * mult		)
         __snake_case:						List[Any]			  =						dim_out if dim_out is not None else dim

         if activation_fn == "gelu":
           __snake_case:						Optional[Any]			  =						GELU(A							,    A		)
         if activation_fn == "gelu-approximate":
           __snake_case:						Any			  =						GELU(A							,    A							,    approximate="""tanh"""		)
         elif activation_fn == "geglu":
           __snake_case:						List[Any]			  =						GEGLU(A							,    A		)
         elif activation_fn == "geglu-approximate":
           __snake_case:						Dict			  =						ApproximateGELU(A							,    A		)

         __snake_case:						Optional[Any]			  =						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 							UpperCAmelCase__			(  self			:   str							,    A			:   Tuple		):
         for module in self.net:
           __snake_case:						List[str]			  =						module(A		)
         return hidden_states







class    __snake_case	(							nn.Module      ):





       '''simple docstring'''







       def __init__(  self			:   Union[str, Any]							,    A			:   int							,    A			:   int							,    A			:   str = "none"		):
         super().__init__()
         __snake_case:						Union[str, Any]			  =						nn.Linear(A							,    A		)
         __snake_case:						Tuple			  =						approximate




       def 							UpperCAmelCase__			(  self			:   List[str]							,    A			:   List[Any]		):
         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 							UpperCAmelCase__			(  self			:   Union[str, Any]							,    A			:   Any		):
         __snake_case:						str			  =						self.proj(A		)
         __snake_case:						Dict			  =						self.gelu(A		)
         return hidden_states







class    __snake_case	(							nn.Module      ):





       '''simple docstring'''







       def __init__(  self			:   Union[str, Any]							,    A			:   int							,    A			:   int		):
         super().__init__()
         __snake_case:						Union[str, Any]			  =						nn.Linear(A							,    dim_out * 2		)




       def 							UpperCAmelCase__			(  self			:   str							,    A			:   Dict		):
         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 							UpperCAmelCase__			(  self			:   Optional[int]							,    A			:   str		):
         __snake_case  ,       __snake_case:						Tuple			  =						self.proj(A		).chunk(2							,    dim=-1		)
         return hidden_states * self.gelu(A		)







class    __snake_case	(							nn.Module      ):





       '''simple docstring'''







       def __init__(  self			:   Tuple							,    A			:   int							,    A			:   int		):
         super().__init__()
         __snake_case:						Optional[int]			  =						nn.Linear(A							,    A		)






       def 							UpperCAmelCase__			(  self			:   Optional[int]							,    A			:   List[Any]		):
         __snake_case:						Any			  =						self.proj(A		)
         return x * torch.sigmoid(1.702 * x		)







class    __snake_case	(							nn.Module      ):





       '''simple docstring'''







       def __init__(  self			:   str							,    A			:   Optional[int]							,    A			:   str		):
         super().__init__()
         __snake_case:						Dict			  =						nn.Embedding(A							,    A		)
         __snake_case:						List[str]			  =						nn.SiLU()
         __snake_case:						List[str]			  =						nn.Linear(A							,    embedding_dim * 2		)
         __snake_case:						Any			  =						nn.LayerNorm(A							,    elementwise_affine=A		)






       def 							UpperCAmelCase__			(  self			:   Any							,    A			:   List[str]							,    A			:   List[str]		):
         __snake_case:						Tuple			  =						self.linear(self.silu(self.emb(A		)		)		)
         __snake_case  ,       __snake_case:						Optional[int]			  =						torch.chunk(A							,    2		)
         __snake_case:						int			  =						self.norm(A		) * (1 + scale) + shift
         return x







class    __snake_case	(							nn.Module      ):





       '''simple docstring'''







       def __init__(  self			:   Union[str, Any]							,    A			:   Optional[int]							,    A			:   List[Any]		):
         super().__init__()

         __snake_case:						str			  =						CombinedTimestepLabelEmbeddings(A							,    A		)

         __snake_case:						List[str]			  =						nn.SiLU()
         __snake_case:						int			  =						nn.Linear(A							,    6 * embedding_dim							,    bias=A		)
         __snake_case:						Any			  =						nn.LayerNorm(A							,    elementwise_affine=A							,    eps=1E-6		)






       def 							UpperCAmelCase__			(  self			:   int							,    A			:   Dict							,    A			:   str							,    A			:   Any							,    A			:   Dict=None		):
         __snake_case:						Dict			  =						self.linear(self.silu(self.emb(A							,    A							,    hidden_dtype=A		)		)		)
         __snake_case  ,       __snake_case  ,       __snake_case  ,       __snake_case  ,       __snake_case  ,       __snake_case:						int			  =						emb.chunk(6							,    dim=1		)
         __snake_case:						str			  =						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      ):





       '''simple docstring'''







       def __init__(  self			:   str							,    A			:   int							,    A			:   int							,    A			:   int							,    A			:   Optional[str] = None							,    A			:   float = 1E-5		):
         super().__init__()
         __snake_case:						List[Any]			  =						num_groups
         __snake_case:						str			  =						eps

         if act_fn is None:
           __snake_case:						List[Any]			  =						None
         else:
           __snake_case:						int			  =						get_activation(A		)

         __snake_case:						Optional[int]			  =						nn.Linear(A							,    out_dim * 2		)






       def 							UpperCAmelCase__			(  self			:   Optional[Any]							,    A			:   str							,    A			:   str		):
         if self.act:
           __snake_case:						List[str]			  =						self.act(A		)
         __snake_case:						Dict			  =						self.linear(A		)
         __snake_case:						Union[str, Any]			  =						emb[:, :, None, None]
         __snake_case  ,       __snake_case:						int			  =						emb.chunk(2							,    dim=1		)

         __snake_case:						Optional[int]			  =						F.group_norm(A							,    self.num_groups							,    eps=self.eps		)
         __snake_case:						List[Any]			  =						x * (1 + scale) + shift
         return x













import warnings

from diffusers import StableDiffusionImgaImgPipeline  # noqa F401


warnings.warn(
    "The `image_to_image.py` script is outdated. Please use directly `from diffusers import"
    " StableDiffusionImg2ImgPipeline` instead."
)






