


"""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:						int							,__lowerCAmelCase:						int							,__lowerCAmelCase:						int							,__lowerCAmelCase:						int							,__lowerCAmelCase:						Optional[Any]=0.0							,__lowerCAmelCase:						Optional[int] = None							,__lowerCAmelCase:						str = "geglu"							,__lowerCAmelCase:						Optional[int] = None							,__lowerCAmelCase:						bool = False							,__lowerCAmelCase:						bool = False							,__lowerCAmelCase:						bool = False							,__lowerCAmelCase:						bool = False							,__lowerCAmelCase:						bool = True							,__lowerCAmelCase:						str = "layer_norm"							,__lowerCAmelCase:						bool = False							,):







     '''simple docstring'''







     super().__init__()
     _lowerCamelCase   :				int							=     only_cross_attention

     _lowerCamelCase   :				Tuple							=     (num_embeds_ada_norm is not None) and norm_type == "ada_norm_zero"
     _lowerCamelCase   :				Union[str, Any]							=     (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   :				Tuple							=     AdaLayerNorm(__lowerCAmelCase							,__lowerCAmelCase    )
     elif self.use_ada_layer_norm_zero:
        _lowerCamelCase   :				Union[str, Any]							=     AdaLayerNormZero(__lowerCAmelCase							,__lowerCAmelCase    )
     else:
        _lowerCamelCase   :				Dict							=     nn.LayerNorm(__lowerCAmelCase							,elementwise_affine=__lowerCAmelCase    )
     _lowerCamelCase   :				Optional[Any]							=     Attention(
         query_dim=__lowerCAmelCase							,heads=__lowerCAmelCase							,dim_head=__lowerCAmelCase							,dropout=__lowerCAmelCase							,bias=__lowerCAmelCase							,cross_attention_dim=cross_attention_dim if only_cross_attention else None							,upcast_attention=__lowerCAmelCase							,)

     # 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   :				int							=     (
            AdaLayerNorm(__lowerCAmelCase							,__lowerCAmelCase    )
            if self.use_ada_layer_norm
            else nn.LayerNorm(__lowerCAmelCase							,elementwise_affine=__lowerCAmelCase    )
        )
        _lowerCamelCase   :				Union[str, Any]							=     Attention(
            query_dim=__lowerCAmelCase							,cross_attention_dim=cross_attention_dim if not double_self_attention else None							,heads=__lowerCAmelCase							,dim_head=__lowerCAmelCase							,dropout=__lowerCAmelCase							,bias=__lowerCAmelCase							,upcast_attention=__lowerCAmelCase							,)  # is self-attn if encoder_hidden_states is none
     else:
        _lowerCamelCase   :				Tuple							=     None
        _lowerCamelCase   :				Any							=     None

     # 3. Feed-forward
     _lowerCamelCase   :				Tuple							=     nn.LayerNorm(__lowerCAmelCase							,elementwise_affine=__lowerCAmelCase    )
     _lowerCamelCase   :				Any							=     FeedForward(__lowerCAmelCase							,dropout=__lowerCAmelCase							,activation_fn=__lowerCAmelCase							,final_dropout=__lowerCAmelCase    )

     # let chunk size default to None
     _lowerCamelCase   :				int							=     None
     _lowerCamelCase   :				Union[str, Any]							=     0


  def 							_lowercase					(			self:						int							,__lowerCAmelCase:						Optional[int]							,__lowerCAmelCase:						int    ):







     '''simple docstring'''







     _lowerCamelCase   :				Any							=     chunk_size
     _lowerCamelCase   :				Dict							=     dim






  def 							_lowercase					(			self:						Tuple							,__lowerCAmelCase:						torch.FloatTensor							,__lowerCAmelCase:						Optional[torch.FloatTensor] = None							,__lowerCAmelCase:						Optional[torch.FloatTensor] = None							,__lowerCAmelCase:						Optional[torch.FloatTensor] = None							,__lowerCAmelCase:						Optional[torch.LongTensor] = None							,__lowerCAmelCase:						Dict[str, Any] = None							,__lowerCAmelCase:						Optional[torch.LongTensor] = None							,):







     '''simple docstring'''







     if self.use_ada_layer_norm:
        _lowerCamelCase   :				List[str]							=     self.norma(__lowerCAmelCase							,__lowerCAmelCase    )
     elif self.use_ada_layer_norm_zero:
        _lowerCamelCase,					_lowerCamelCase,					_lowerCamelCase,					_lowerCamelCase,					_lowerCamelCase   :				Optional[Any]							=     self.norma(
            __lowerCAmelCase							,__lowerCAmelCase							,__lowerCAmelCase							,hidden_dtype=hidden_states.dtype    )
     else:
        _lowerCamelCase   :				List[str]							=     self.norma(__lowerCAmelCase    )

     _lowerCamelCase   :				Any							=     cross_attention_kwargs if cross_attention_kwargs is not None else {}

     _lowerCamelCase   :				str							=     self.attna(
         __lowerCAmelCase							,encoder_hidden_states=encoder_hidden_states if self.only_cross_attention else None							,attention_mask=__lowerCAmelCase							,**__lowerCAmelCase							,)
     if self.use_ada_layer_norm_zero:
        _lowerCamelCase   :				List[Any]							=     gate_msa.unsqueeze(1    ) * attn_output
     _lowerCamelCase   :				str							=     attn_output + hidden_states

     # 2. Cross-Attention
     if self.attna is not None:
        _lowerCamelCase   :				Optional[int]							=     (
            self.norma(__lowerCAmelCase							,__lowerCAmelCase    ) if self.use_ada_layer_norm else self.norma(__lowerCAmelCase    )
        )

        _lowerCamelCase   :				Optional[int]							=     self.attna(
            __lowerCAmelCase							,encoder_hidden_states=__lowerCAmelCase							,attention_mask=__lowerCAmelCase							,**__lowerCAmelCase							,)
        _lowerCamelCase   :				List[Any]							=     attn_output + hidden_states

     # 3. Feed-forward
     _lowerCamelCase   :				int							=     self.norma(__lowerCAmelCase    )

     if self.use_ada_layer_norm_zero:
        _lowerCamelCase   :				Optional[Any]							=     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   :				List[Any]							=     norm_hidden_states.shape[self._chunk_dim] // self._chunk_size
        _lowerCamelCase   :				Union[str, Any]							=     torch.cat(
            [self.ff(__lowerCAmelCase    ) for hid_slice in norm_hidden_states.chunk(__lowerCAmelCase							,dim=self._chunk_dim    )]							,dim=self._chunk_dim							,)
     else:
        _lowerCamelCase   :				Any							=     self.ff(__lowerCAmelCase    )

     if self.use_ada_layer_norm_zero:
        _lowerCamelCase   :				Optional[int]							=     gate_mlp.unsqueeze(1    ) * ff_output

     _lowerCamelCase   :				int							=     ff_output + hidden_states

     return hidden_states




class 							A_      (       nn.Module ):


  def __init__(			self:						Any							,__lowerCAmelCase:						int							,__lowerCAmelCase:						Optional[int] = None							,__lowerCAmelCase:						int = 4							,__lowerCAmelCase:						float = 0.0							,__lowerCAmelCase:						str = "geglu"							,__lowerCAmelCase:						bool = False							,):







     '''simple docstring'''







     super().__init__()
     _lowerCamelCase   :				Dict							=     int(dim * mult    )
     _lowerCamelCase   :				str							=     dim_out if dim_out is not None else dim

     if activation_fn == "gelu":
        _lowerCamelCase   :				str							=     GELU(__lowerCAmelCase							,__lowerCAmelCase    )
     if activation_fn == "gelu-approximate":
        _lowerCamelCase   :				Any							=     GELU(__lowerCAmelCase							,__lowerCAmelCase							,approximate="tanh"    )
     elif activation_fn == "geglu":
        _lowerCamelCase   :				Union[str, Any]							=     GEGLU(__lowerCAmelCase							,__lowerCAmelCase    )
     elif activation_fn == "geglu-approximate":
        _lowerCamelCase   :				Optional[int]							=     ApproximateGELU(__lowerCAmelCase							,__lowerCAmelCase    )

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






  def 							_lowercase					(			self:						Union[str, Any]							,__lowerCAmelCase:						int    ):







     '''simple docstring'''







     for module in self.net:
        _lowerCamelCase   :				int							=     module(__lowerCAmelCase    )
     return hidden_states




class 							A_      (       nn.Module ):


  def __init__(			self:						Any							,__lowerCAmelCase:						int							,__lowerCAmelCase:						int							,__lowerCAmelCase:						str = "none"    ):







     '''simple docstring'''







     super().__init__()
     _lowerCamelCase   :				Dict							=     nn.Linear(__lowerCAmelCase							,__lowerCAmelCase    )
     _lowerCamelCase   :				int							=     approximate


  def 							_lowercase					(			self:						Optional[int]							,__lowerCAmelCase:						List[Any]    ):







     '''simple docstring'''







     if gate.device.type != "mps":
        return F.gelu(__lowerCAmelCase							,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 							_lowercase					(			self:						Dict							,__lowerCAmelCase:						Dict    ):







     '''simple docstring'''







     _lowerCamelCase   :				Optional[Any]							=     self.proj(__lowerCAmelCase    )
     _lowerCamelCase   :				List[Any]							=     self.gelu(__lowerCAmelCase    )
     return hidden_states




class 							A_      (       nn.Module ):


  def __init__(			self:						str							,__lowerCAmelCase:						int							,__lowerCAmelCase:						int    ):







     '''simple docstring'''







     super().__init__()
     _lowerCamelCase   :				Dict							=     nn.Linear(__lowerCAmelCase							,dim_out * 2    )


  def 							_lowercase					(			self:						List[str]							,__lowerCAmelCase:						Optional[Any]    ):







     '''simple docstring'''







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






  def 							_lowercase					(			self:						Optional[int]							,__lowerCAmelCase:						str    ):







     '''simple docstring'''







     _lowerCamelCase,					_lowerCamelCase   :				Dict							=     self.proj(__lowerCAmelCase    ).chunk(2							,dim=-1    )
     return hidden_states * self.gelu(__lowerCAmelCase    )




class 							A_      (       nn.Module ):


  def __init__(			self:						Tuple							,__lowerCAmelCase:						int							,__lowerCAmelCase:						int    ):







     '''simple docstring'''







     super().__init__()
     _lowerCamelCase   :				Optional[Any]							=     nn.Linear(__lowerCAmelCase							,__lowerCAmelCase    )






  def 							_lowercase					(			self:						List[Any]							,__lowerCAmelCase:						Union[str, Any]    ):







     '''simple docstring'''







     _lowerCamelCase   :				Optional[Any]							=     self.proj(__lowerCAmelCase    )
     return x * torch.sigmoid(1.7_02 * x    )




class 							A_      (       nn.Module ):


  def __init__(			self:						Optional[int]							,__lowerCAmelCase:						Tuple							,__lowerCAmelCase:						Tuple    ):







     '''simple docstring'''







     super().__init__()
     _lowerCamelCase   :				Dict							=     nn.Embedding(__lowerCAmelCase							,__lowerCAmelCase    )
     _lowerCamelCase   :				Dict							=     nn.SiLU()
     _lowerCamelCase   :				Dict							=     nn.Linear(__lowerCAmelCase							,embedding_dim * 2    )
     _lowerCamelCase   :				Dict							=     nn.LayerNorm(__lowerCAmelCase							,elementwise_affine=__lowerCAmelCase    )






  def 							_lowercase					(			self:						Dict							,__lowerCAmelCase:						List[str]							,__lowerCAmelCase:						Dict    ):







     '''simple docstring'''







     _lowerCamelCase   :				List[Any]							=     self.linear(self.silu(self.emb(__lowerCAmelCase    )    )    )
     _lowerCamelCase,					_lowerCamelCase   :				Tuple							=     torch.chunk(__lowerCAmelCase							,2    )
     _lowerCamelCase   :				Tuple							=     self.norm(__lowerCAmelCase    ) * (1 + scale) + shift
     return x




class 							A_      (       nn.Module ):


  def __init__(			self:						Tuple							,__lowerCAmelCase:						Union[str, Any]							,__lowerCAmelCase:						str    ):







     '''simple docstring'''







     super().__init__()

     _lowerCamelCase   :				Optional[Any]							=     CombinedTimestepLabelEmbeddings(__lowerCAmelCase							,__lowerCAmelCase    )

     _lowerCamelCase   :				List[Any]							=     nn.SiLU()
     _lowerCamelCase   :				Union[str, Any]							=     nn.Linear(__lowerCAmelCase							,6 * embedding_dim							,bias=__lowerCAmelCase    )
     _lowerCamelCase   :				Optional[Any]							=     nn.LayerNorm(__lowerCAmelCase							,elementwise_affine=__lowerCAmelCase							,eps=1e-6    )






  def 							_lowercase					(			self:						Optional[Any]							,__lowerCAmelCase:						Optional[Any]							,__lowerCAmelCase:						Optional[Any]							,__lowerCAmelCase:						Dict							,__lowerCAmelCase:						Tuple=None    ):







     '''simple docstring'''







     _lowerCamelCase   :				Union[str, Any]							=     self.linear(self.silu(self.emb(__lowerCAmelCase							,__lowerCAmelCase							,hidden_dtype=__lowerCAmelCase    )    )    )
     _lowerCamelCase,					_lowerCamelCase,					_lowerCamelCase,					_lowerCamelCase,					_lowerCamelCase,					_lowerCamelCase   :				List[Any]							=     emb.chunk(6							,dim=1    )
     _lowerCamelCase   :				int							=     self.norm(__lowerCAmelCase    ) * (1 + scale_msa[:, None]) + shift_msa[:, None]
     return x, gate_msa, shift_mlp, scale_mlp, gate_mlp




class 							A_      (       nn.Module ):


  def __init__(			self:						List[str]							,__lowerCAmelCase:						int							,__lowerCAmelCase:						int							,__lowerCAmelCase:						int							,__lowerCAmelCase:						Optional[str] = None							,__lowerCAmelCase:						float = 1e-5    ):







     '''simple docstring'''







     super().__init__()
     _lowerCamelCase   :				str							=     num_groups
     _lowerCamelCase   :				int							=     eps

     if act_fn is None:
        _lowerCamelCase   :				List[str]							=     None
     else:
        _lowerCamelCase   :				List[str]							=     get_activation(__lowerCAmelCase    )

     _lowerCamelCase   :				Dict							=     nn.Linear(__lowerCAmelCase							,out_dim * 2    )






  def 							_lowercase					(			self:						Optional[int]							,__lowerCAmelCase:						Any							,__lowerCAmelCase:						Optional[Any]    ):







     '''simple docstring'''







     if self.act:
        _lowerCamelCase   :				Optional[Any]							=     self.act(__lowerCAmelCase    )
     _lowerCamelCase   :				int							=     self.linear(__lowerCAmelCase    )
     _lowerCamelCase   :				int							=     emb[:, :, None, None]
     _lowerCamelCase,					_lowerCamelCase   :				List[Any]							=     emb.chunk(2							,dim=1    )

     _lowerCamelCase   :				List[str]							=     F.group_norm(__lowerCAmelCase							,self.num_groups							,eps=self.eps    )
     _lowerCamelCase   :				List[str]							=     x * (1 + scale) + shift
     return x

import unittest

import numpy as np
import torch

from diffusers import PNDMPipeline, PNDMScheduler, UNetaDModel
from diffusers.utils.testing_utils import enable_full_determinism, require_torch, slow, torch_device


enable_full_determinism()







class 							lowerCAmelCase__   ( unittest.TestCase						):

						@property
						def 			A_					(      self					)       ->   int:


								'''simple docstring'''

								torch.manual_seed(0					)
								_UpperCamelCase         =						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 			A_					(      self					)       ->   Any:


								'''simple docstring'''

								_UpperCamelCase         =						self.dummy_uncond_unet
								_UpperCamelCase         =						PNDMScheduler()

								_UpperCamelCase         =						PNDMPipeline(unet=a							,				scheduler=a					)
								pndm.to(a					)
								pndm.set_progress_bar_config(disable=a					)

								_UpperCamelCase         =						torch.manual_seed(0					)
								_UpperCamelCase         =						pndm(generator=a							,				num_inference_steps=20							,				output_type="""numpy"""					).images

								_UpperCamelCase         =						torch.manual_seed(0					)
								_UpperCamelCase         =						pndm(generator=a							,				num_inference_steps=20							,				output_type="""numpy"""							,				return_dict=a					)[0]

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

								assert image.shape == (1, 32, 32, 3)
								_UpperCamelCase         =						np.array([1.0, 1.0, 0.0, 1.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 							lowerCAmelCase__   ( unittest.TestCase						):

						def 			A_					(      self					)       ->   Optional[Any]:


								'''simple docstring'''

								_UpperCamelCase         =						"""google/ddpm-cifar10-32"""

								_UpperCamelCase         =						UNetaDModel.from_pretrained(a					)
								_UpperCamelCase         =						PNDMScheduler()

								_UpperCamelCase         =						PNDMPipeline(unet=a							,				scheduler=a					)
								pndm.to(a					)
								pndm.set_progress_bar_config(disable=a					)
								_UpperCamelCase         =						torch.manual_seed(0					)
								_UpperCamelCase         =						pndm(generator=a							,				output_type="""numpy"""					).images

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

								assert image.shape == (1, 32, 32, 3)
								_UpperCamelCase         =						np.array([0.1564, 0.1_4645, 0.1406, 0.1_4715, 0.1_2425, 0.1_4045, 0.1_3115, 0.1_2175, 0.125]					)

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



