






"""simple docstring"""





from dataclasses import dataclass
from typing import Optional

import numpy as np
import torch
import torch.nn as nn

from ..utils import BaseOutput, is_torch_version, randn_tensor
from .attention_processor import SpatialNorm
from .unet_ad_blocks import UNetMidBlockaD, get_down_block, get_up_block






@dataclass
class 	a		(   __snake_case						):
    SCREAMING_SNAKE_CASE		:	torch.FloatTensor






class 	a		(   nn.Module						):






    def __init__(			self	:				Any			,		__SCREAMING_SNAKE_CASE	:				List[Any]=3			,		__SCREAMING_SNAKE_CASE	:				Union[str, Any]=3			,		__SCREAMING_SNAKE_CASE	:				Dict=("DownEncoderBlock2D",)			,		__SCREAMING_SNAKE_CASE	:				Optional[Any]=(64,)			,		__SCREAMING_SNAKE_CASE	:				Union[str, Any]=2			,		__SCREAMING_SNAKE_CASE	:				Optional[int]=32			,		__SCREAMING_SNAKE_CASE	:				Optional[Any]="silu"			,		__SCREAMING_SNAKE_CASE	:				Tuple=True			,		)	->						Dict:
      super().__init__()
      lowerCamelCase_       	=		layers_per_block

      lowerCamelCase_       	=		torch.nn.Convad(
          __SCREAMING_SNAKE_CASE			,		block_out_channels[0]			,		kernel_size=3			,		stride=1			,		padding=1			,		)

      lowerCamelCase_       	=		None
      lowerCamelCase_       	=		nn.ModuleList([]   )

      # down
      lowerCamelCase_       	=		block_out_channels[0]
      for i, down_block_type in enumerate(__SCREAMING_SNAKE_CASE   ):
        lowerCamelCase_       	=		output_channel
        lowerCamelCase_       	=		block_out_channels[i]
        lowerCamelCase_       	=		i == len(__SCREAMING_SNAKE_CASE   ) - 1

        lowerCamelCase_       	=		get_down_block(
            __SCREAMING_SNAKE_CASE			,		num_layers=self.layers_per_block			,		in_channels=__SCREAMING_SNAKE_CASE			,		out_channels=__SCREAMING_SNAKE_CASE			,		add_downsample=not is_final_block			,		resnet_eps=1e-6			,		downsample_padding=0			,		resnet_act_fn=__SCREAMING_SNAKE_CASE			,		resnet_groups=__SCREAMING_SNAKE_CASE			,		attention_head_dim=__SCREAMING_SNAKE_CASE			,		temb_channels=__SCREAMING_SNAKE_CASE			,		)
        self.down_blocks.append(__SCREAMING_SNAKE_CASE   )

      # mid
      lowerCamelCase_       	=		UNetMidBlockaD(
          in_channels=block_out_channels[-1]			,		resnet_eps=1e-6			,		resnet_act_fn=__SCREAMING_SNAKE_CASE			,		output_scale_factor=1			,		resnet_time_scale_shift='default'			,		attention_head_dim=block_out_channels[-1]			,		resnet_groups=__SCREAMING_SNAKE_CASE			,		temb_channels=__SCREAMING_SNAKE_CASE			,		)

      # out
      lowerCamelCase_       	=		nn.GroupNorm(num_channels=block_out_channels[-1]			,		num_groups=__SCREAMING_SNAKE_CASE			,		eps=1e-6   )
      lowerCamelCase_       	=		nn.SiLU()

      lowerCamelCase_       	=		2 * out_channels if double_z else out_channels
      lowerCamelCase_       	=		nn.Convad(block_out_channels[-1]			,		__SCREAMING_SNAKE_CASE			,		3			,		padding=1   )

      lowerCamelCase_       	=		False






    def   UpperCamelCase			(			self	:				int			,		__SCREAMING_SNAKE_CASE	:				Tuple   )	->						Union[str, Any]:
      lowerCamelCase_       	=		x
      lowerCamelCase_       	=		self.conv_in(__SCREAMING_SNAKE_CASE   )

      if self.training and self.gradient_checkpointing:

        def create_custom_forward(__SCREAMING_SNAKE_CASE	:				int   ):
          def custom_forward(*__SCREAMING_SNAKE_CASE	:				List[str]   ):
            return module(*__SCREAMING_SNAKE_CASE   )

          return custom_forward

        # down
        if is_torch_version('>='			,		'1.11.0'   ):
          for down_block in self.down_blocks:
            lowerCamelCase_       	=		torch.utils.checkpoint.checkpoint(
                create_custom_forward(__SCREAMING_SNAKE_CASE   )			,		__SCREAMING_SNAKE_CASE			,		use_reentrant=__SCREAMING_SNAKE_CASE   )
          # middle
          lowerCamelCase_       	=		torch.utils.checkpoint.checkpoint(
              create_custom_forward(self.mid_block   )			,		__SCREAMING_SNAKE_CASE			,		use_reentrant=__SCREAMING_SNAKE_CASE   )
        else:
          for down_block in self.down_blocks:
            lowerCamelCase_       	=		torch.utils.checkpoint.checkpoint(create_custom_forward(__SCREAMING_SNAKE_CASE   )			,		__SCREAMING_SNAKE_CASE   )
          # middle
          lowerCamelCase_       	=		torch.utils.checkpoint.checkpoint(create_custom_forward(self.mid_block   )			,		__SCREAMING_SNAKE_CASE   )

      else:
        # down
        for down_block in self.down_blocks:
          lowerCamelCase_       	=		down_block(__SCREAMING_SNAKE_CASE   )

        # middle
        lowerCamelCase_       	=		self.mid_block(__SCREAMING_SNAKE_CASE   )

      # post-process
      lowerCamelCase_       	=		self.conv_norm_out(__SCREAMING_SNAKE_CASE   )
      lowerCamelCase_       	=		self.conv_act(__SCREAMING_SNAKE_CASE   )
      lowerCamelCase_       	=		self.conv_out(__SCREAMING_SNAKE_CASE   )

      return sample






class 	a		(   nn.Module						):






    def __init__(			self	:				List[str]			,		__SCREAMING_SNAKE_CASE	:				Any=3			,		__SCREAMING_SNAKE_CASE	:				Any=3			,		__SCREAMING_SNAKE_CASE	:				Any=("UpDecoderBlock2D",)			,		__SCREAMING_SNAKE_CASE	:				Optional[Any]=(64,)			,		__SCREAMING_SNAKE_CASE	:				int=2			,		__SCREAMING_SNAKE_CASE	:				Any=32			,		__SCREAMING_SNAKE_CASE	:				Optional[Any]="silu"			,		__SCREAMING_SNAKE_CASE	:				Tuple="group"			,		)	->						List[str]:
      super().__init__()
      lowerCamelCase_       	=		layers_per_block

      lowerCamelCase_       	=		nn.Convad(
          __SCREAMING_SNAKE_CASE			,		block_out_channels[-1]			,		kernel_size=3			,		stride=1			,		padding=1			,		)

      lowerCamelCase_       	=		None
      lowerCamelCase_       	=		nn.ModuleList([]   )

      lowerCamelCase_       	=		in_channels if norm_type == 'spatial' else None

      # mid
      lowerCamelCase_       	=		UNetMidBlockaD(
          in_channels=block_out_channels[-1]			,		resnet_eps=1e-6			,		resnet_act_fn=__SCREAMING_SNAKE_CASE			,		output_scale_factor=1			,		resnet_time_scale_shift='default' if norm_type == 'group' else norm_type			,		attention_head_dim=block_out_channels[-1]			,		resnet_groups=__SCREAMING_SNAKE_CASE			,		temb_channels=__SCREAMING_SNAKE_CASE			,		)

      # up
      lowerCamelCase_       	=		list(reversed(__SCREAMING_SNAKE_CASE   )   )
      lowerCamelCase_       	=		reversed_block_out_channels[0]
      for i, up_block_type in enumerate(__SCREAMING_SNAKE_CASE   ):
        lowerCamelCase_       	=		output_channel
        lowerCamelCase_       	=		reversed_block_out_channels[i]

        lowerCamelCase_       	=		i == len(__SCREAMING_SNAKE_CASE   ) - 1

        lowerCamelCase_       	=		get_up_block(
            __SCREAMING_SNAKE_CASE			,		num_layers=self.layers_per_block + 1			,		in_channels=__SCREAMING_SNAKE_CASE			,		out_channels=__SCREAMING_SNAKE_CASE			,		prev_output_channel=__SCREAMING_SNAKE_CASE			,		add_upsample=not is_final_block			,		resnet_eps=1e-6			,		resnet_act_fn=__SCREAMING_SNAKE_CASE			,		resnet_groups=__SCREAMING_SNAKE_CASE			,		attention_head_dim=__SCREAMING_SNAKE_CASE			,		temb_channels=__SCREAMING_SNAKE_CASE			,		resnet_time_scale_shift=__SCREAMING_SNAKE_CASE			,		)
        self.up_blocks.append(__SCREAMING_SNAKE_CASE   )
        lowerCamelCase_       	=		output_channel

      # out
      if norm_type == "spatial":
        lowerCamelCase_       	=		SpatialNorm(block_out_channels[0]			,		__SCREAMING_SNAKE_CASE   )
      else:
        lowerCamelCase_       	=		nn.GroupNorm(num_channels=block_out_channels[0]			,		num_groups=__SCREAMING_SNAKE_CASE			,		eps=1e-6   )
      lowerCamelCase_       	=		nn.SiLU()
      lowerCamelCase_       	=		nn.Convad(block_out_channels[0]			,		__SCREAMING_SNAKE_CASE			,		3			,		padding=1   )

      lowerCamelCase_       	=		False






    def   UpperCamelCase			(			self	:				Optional[int]			,		__SCREAMING_SNAKE_CASE	:				Optional[int]			,		__SCREAMING_SNAKE_CASE	:				Any=None   )	->						Optional[int]:
      lowerCamelCase_       	=		z
      lowerCamelCase_       	=		self.conv_in(__SCREAMING_SNAKE_CASE   )

      lowerCamelCase_       	=		next(iter(self.up_blocks.parameters()   )   ).dtype
      if self.training and self.gradient_checkpointing:

        def create_custom_forward(__SCREAMING_SNAKE_CASE	:				Optional[int]   ):
          def custom_forward(*__SCREAMING_SNAKE_CASE	:				int   ):
            return module(*__SCREAMING_SNAKE_CASE   )

          return custom_forward

        if is_torch_version('>='			,		'1.11.0'   ):
          # middle
          lowerCamelCase_       	=		torch.utils.checkpoint.checkpoint(
              create_custom_forward(self.mid_block   )			,		__SCREAMING_SNAKE_CASE			,		__SCREAMING_SNAKE_CASE			,		use_reentrant=__SCREAMING_SNAKE_CASE   )
          lowerCamelCase_       	=		sample.to(__SCREAMING_SNAKE_CASE   )

          # up
          for up_block in self.up_blocks:
            lowerCamelCase_       	=		torch.utils.checkpoint.checkpoint(
                create_custom_forward(__SCREAMING_SNAKE_CASE   )			,		__SCREAMING_SNAKE_CASE			,		__SCREAMING_SNAKE_CASE			,		use_reentrant=__SCREAMING_SNAKE_CASE   )
        else:
          # middle
          lowerCamelCase_       	=		torch.utils.checkpoint.checkpoint(
              create_custom_forward(self.mid_block   )			,		__SCREAMING_SNAKE_CASE			,		__SCREAMING_SNAKE_CASE   )
          lowerCamelCase_       	=		sample.to(__SCREAMING_SNAKE_CASE   )

          # up
          for up_block in self.up_blocks:
            lowerCamelCase_       	=		torch.utils.checkpoint.checkpoint(create_custom_forward(__SCREAMING_SNAKE_CASE   )			,		__SCREAMING_SNAKE_CASE			,		__SCREAMING_SNAKE_CASE   )
      else:
        # middle
        lowerCamelCase_       	=		self.mid_block(__SCREAMING_SNAKE_CASE			,		__SCREAMING_SNAKE_CASE   )
        lowerCamelCase_       	=		sample.to(__SCREAMING_SNAKE_CASE   )

        # up
        for up_block in self.up_blocks:
          lowerCamelCase_       	=		up_block(__SCREAMING_SNAKE_CASE			,		__SCREAMING_SNAKE_CASE   )

        # post-process
      if latent_embeds is None:
        lowerCamelCase_       	=		self.conv_norm_out(__SCREAMING_SNAKE_CASE   )
      else:
        lowerCamelCase_       	=		self.conv_norm_out(__SCREAMING_SNAKE_CASE			,		__SCREAMING_SNAKE_CASE   )
      lowerCamelCase_       	=		self.conv_act(__SCREAMING_SNAKE_CASE   )
      lowerCamelCase_       	=		self.conv_out(__SCREAMING_SNAKE_CASE   )

      return sample






class 	a		(   nn.Module						):






    def __init__(			self	:				Union[str, Any]			,		__SCREAMING_SNAKE_CASE	:				Optional[int]			,		__SCREAMING_SNAKE_CASE	:				Union[str, Any]			,		__SCREAMING_SNAKE_CASE	:				Optional[Any]			,		__SCREAMING_SNAKE_CASE	:				List[Any]=None			,		__SCREAMING_SNAKE_CASE	:				Any="random"			,		__SCREAMING_SNAKE_CASE	:				Optional[Any]=False			,		__SCREAMING_SNAKE_CASE	:				List[Any]=True   )	->						Optional[Any]:
      super().__init__()
      lowerCamelCase_       	=		n_e
      lowerCamelCase_       	=		vq_embed_dim
      lowerCamelCase_       	=		beta
      lowerCamelCase_       	=		legacy

      lowerCamelCase_       	=		nn.Embedding(self.n_e			,		self.vq_embed_dim   )
      self.embedding.weight.data.uniform_(-1.0 / self.n_e			,		1.0 / self.n_e   )

      lowerCamelCase_       	=		remap
      if self.remap is not None:
        self.register_buffer('used'			,		torch.tensor(np.load(self.remap   )   )   )
        lowerCamelCase_       	=		self.used.shape[0]
        lowerCamelCase_       	=		unknown_index  # "random" or "extra" or integer
        if self.unknown_index == "extra":
          lowerCamelCase_       	=		self.re_embed
          lowerCamelCase_       	=		self.re_embed + 1
        print(
            F'''Remapping {self.n_e} indices to {self.re_embed} indices. '''
            F'''Using {self.unknown_index} for unknown indices.'''   )
      else:
        lowerCamelCase_       	=		n_e

      lowerCamelCase_       	=		sane_index_shape






    def   UpperCamelCase			(			self	:				Optional[Any]			,		__SCREAMING_SNAKE_CASE	:				Union[str, Any]   )	->						List[str]:
      lowerCamelCase_       	=		inds.shape
      assert len(__SCREAMING_SNAKE_CASE   ) > 1
      lowerCamelCase_       	=		inds.reshape(ishape[0]			,		-1   )
      lowerCamelCase_       	=		self.used.to(__SCREAMING_SNAKE_CASE   )
      lowerCamelCase_       	=		(inds[:, :, None] == used[None, None, ...]).long()
      lowerCamelCase_       	=		match.argmax(-1   )
      lowerCamelCase_       	=		match.sum(2   ) < 1
      if self.unknown_index == "random":
        lowerCamelCase_       	=		torch.randint(0			,		self.re_embed			,		size=new[unknown].shape   ).to(device=new.device   )
      else:
        lowerCamelCase_       	=		self.unknown_index
      return new.reshape(__SCREAMING_SNAKE_CASE   )






    def   UpperCamelCase			(			self	:				Optional[int]			,		__SCREAMING_SNAKE_CASE	:				Optional[Any]   )	->						Tuple:
      lowerCamelCase_       	=		inds.shape
      assert len(__SCREAMING_SNAKE_CASE   ) > 1
      lowerCamelCase_       	=		inds.reshape(ishape[0]			,		-1   )
      lowerCamelCase_       	=		self.used.to(__SCREAMING_SNAKE_CASE   )
      if self.re_embed > self.used.shape[0]:  # extra token
        lowerCamelCase_       	=		0  # simply set to zero
      lowerCamelCase_       	=		torch.gather(used[None, :][inds.shape[0] * [0], :]			,		1			,		__SCREAMING_SNAKE_CASE   )
      return back.reshape(__SCREAMING_SNAKE_CASE   )






    def   UpperCamelCase			(			self	:				Tuple			,		__SCREAMING_SNAKE_CASE	:				int   )	->						Dict:
      # reshape z -> (batch, height, width, channel) and flatten
      lowerCamelCase_       	=		z.permute(0			,		2			,		3			,		1   ).contiguous()
      lowerCamelCase_       	=		z.view(-1			,		self.vq_embed_dim   )

      # distances from z to embeddings e_j (z - e)^2 = z^2 + e^2 - 2 e * z
      lowerCamelCase_       	=		torch.argmin(torch.cdist(__SCREAMING_SNAKE_CASE			,		self.embedding.weight   )			,		dim=1   )

      lowerCamelCase_       	=		self.embedding(__SCREAMING_SNAKE_CASE   ).view(z.shape   )
      lowerCamelCase_       	=		None
      lowerCamelCase_       	=		None

      # compute loss for embedding
      if not self.legacy:
        lowerCamelCase_       	=		self.beta * torch.mean((z_q.detach() - z) ** 2   ) + torch.mean((z_q - z.detach()) ** 2   )
      else:
        lowerCamelCase_       	=		torch.mean((z_q.detach() - z) ** 2   ) + self.beta * torch.mean((z_q - z.detach()) ** 2   )

      # preserve gradients
      lowerCamelCase_       	=		z + (z_q - z).detach()

      # reshape back to match original input shape
      lowerCamelCase_       	=		z_q.permute(0			,		3			,		1			,		2   ).contiguous()

      if self.remap is not None:
        lowerCamelCase_       	=		min_encoding_indices.reshape(z.shape[0]			,		-1   )  # add batch axis
        lowerCamelCase_       	=		self.remap_to_used(__SCREAMING_SNAKE_CASE   )
        lowerCamelCase_       	=		min_encoding_indices.reshape(-1			,		1   )  # flatten

      if self.sane_index_shape:
        lowerCamelCase_       	=		min_encoding_indices.reshape(z_q.shape[0]			,		z_q.shape[2]			,		z_q.shape[3]   )

      return z_q, loss, (perplexity, min_encodings, min_encoding_indices)






    def   UpperCamelCase			(			self	:				List[str]			,		__SCREAMING_SNAKE_CASE	:				Any			,		__SCREAMING_SNAKE_CASE	:				Optional[Any]   )	->						Optional[int]:
      # shape specifying (batch, height, width, channel)
      if self.remap is not None:
        lowerCamelCase_       	=		indices.reshape(shape[0]			,		-1   )  # add batch axis
        lowerCamelCase_       	=		self.unmap_to_all(__SCREAMING_SNAKE_CASE   )
        lowerCamelCase_       	=		indices.reshape(-1   )  # flatten again

      # get quantized latent vectors
      lowerCamelCase_       	=		self.embedding(__SCREAMING_SNAKE_CASE   )

      if shape is not None:
        lowerCamelCase_       	=		z_q.view(__SCREAMING_SNAKE_CASE   )
        # reshape back to match original input shape
        lowerCamelCase_       	=		z_q.permute(0			,		3			,		1			,		2   ).contiguous()

      return z_q






class 	a		(   __snake_case						):






    def __init__(			self	:				Union[str, Any]			,		__SCREAMING_SNAKE_CASE	:				List[Any]			,		__SCREAMING_SNAKE_CASE	:				str=False   )	->						List[str]:
      lowerCamelCase_       	=		parameters
      lowerCamelCase_	,      lowerCamelCase_       	=		torch.chunk(__SCREAMING_SNAKE_CASE			,		2			,		dim=1   )
      lowerCamelCase_       	=		torch.clamp(self.logvar			,		-30.0			,		20.0   )
      lowerCamelCase_       	=		deterministic
      lowerCamelCase_       	=		torch.exp(0.5 * self.logvar   )
      lowerCamelCase_       	=		torch.exp(self.logvar   )
      if self.deterministic:
        lowerCamelCase_       	=		lowerCamelCase_       	=		torch.zeros_like(
            self.mean			,		device=self.parameters.device			,		dtype=self.parameters.dtype   )






    def   UpperCamelCase			(			self	:				Any			,		__SCREAMING_SNAKE_CASE	:				Optional[torch.Generator] = None   )	->						torch.FloatTensor:
      # make sure sample is on the same device as the parameters and has same dtype
      lowerCamelCase_       	=		randn_tensor(
          self.mean.shape			,		generator=__SCREAMING_SNAKE_CASE			,		device=self.parameters.device			,		dtype=self.parameters.dtype   )
      lowerCamelCase_       	=		self.mean + self.std * sample
      return x






    def   UpperCamelCase			(			self	:				List[str]			,		__SCREAMING_SNAKE_CASE	:				Tuple=None   )	->						Dict:
      if self.deterministic:
        return torch.Tensor([0.0]   )
      else:
        if other is None:
          return 0.5 * torch.sum(torch.pow(self.mean			,		2   ) + self.var - 1.0 - self.logvar			,		dim=[1, 2, 3]   )
        else:
          return 0.5 * torch.sum(
              torch.pow(self.mean - other.mean			,		2   ) / other.var
              + self.var / other.var
              - 1.0
              - self.logvar
              + other.logvar			,		dim=[1, 2, 3]			,		)






    def   UpperCamelCase			(			self	:				int			,		__SCREAMING_SNAKE_CASE	:				Tuple			,		__SCREAMING_SNAKE_CASE	:				List[str]=[1, 2, 3]   )	->						Any:
      if self.deterministic:
        return torch.Tensor([0.0]   )
      lowerCamelCase_       	=		np.log(2.0 * np.pi   )
      return 0.5 * torch.sum(logtwopi + self.logvar + torch.pow(sample - self.mean			,		2   ) / self.var			,		dim=__SCREAMING_SNAKE_CASE   )






    def   UpperCamelCase			(			self	:				Union[str, Any]   )	->						Dict:
      return self.mean












"""simple docstring"""





from dataclasses import dataclass
from typing import Dict, Optional, Union

import torch
import torch.nn.functional as F
from torch import nn

from ..configuration_utils import ConfigMixin, register_to_config
from ..utils import BaseOutput
from .attention import BasicTransformerBlock
from .attention_processor import AttentionProcessor, AttnProcessor
from .embeddings import TimestepEmbedding, Timesteps
from .modeling_utils import ModelMixin






@dataclass
class 	a		(   __snake_case						):
    SCREAMING_SNAKE_CASE		:	torch.FloatTensor






class 	a		(   __snake_case    , __snake_case						):






    @register_to_config
    def __init__(			self	:				Any			,		__SCREAMING_SNAKE_CASE	:				int = 32			,		__SCREAMING_SNAKE_CASE	:				int = 64			,		__SCREAMING_SNAKE_CASE	:				int = 20			,		__SCREAMING_SNAKE_CASE	:				int = 768			,		__SCREAMING_SNAKE_CASE	:				int=77			,		__SCREAMING_SNAKE_CASE	:				Any=4			,		__SCREAMING_SNAKE_CASE	:				float = 0.0			,		__SCREAMING_SNAKE_CASE	:				str = "silu"			,		__SCREAMING_SNAKE_CASE	:				Optional[str] = None			,		__SCREAMING_SNAKE_CASE	:				Optional[str] = None			,		__SCREAMING_SNAKE_CASE	:				Optional[str] = "linear"			,		__SCREAMING_SNAKE_CASE	:				Optional[str] = "prd"			,		__SCREAMING_SNAKE_CASE	:				Optional[int] = None			,		__SCREAMING_SNAKE_CASE	:				Optional[int] = None			,		__SCREAMING_SNAKE_CASE	:				Optional[int] = None			,		)	->						List[Any]:
      super().__init__()
      lowerCamelCase_       	=		num_attention_heads
      lowerCamelCase_       	=		attention_head_dim
      lowerCamelCase_       	=		num_attention_heads * attention_head_dim
      lowerCamelCase_       	=		additional_embeddings

      lowerCamelCase_       	=		time_embed_dim or inner_dim
      lowerCamelCase_       	=		embedding_proj_dim or embedding_dim
      lowerCamelCase_       	=		clip_embed_dim or embedding_dim

      lowerCamelCase_       	=		Timesteps(__SCREAMING_SNAKE_CASE			,		__SCREAMING_SNAKE_CASE			,		0   )
      lowerCamelCase_       	=		TimestepEmbedding(__SCREAMING_SNAKE_CASE			,		__SCREAMING_SNAKE_CASE			,		out_dim=__SCREAMING_SNAKE_CASE			,		act_fn=__SCREAMING_SNAKE_CASE   )

      lowerCamelCase_       	=		nn.Linear(__SCREAMING_SNAKE_CASE			,		__SCREAMING_SNAKE_CASE   )

      if embedding_proj_norm_type is None:
        lowerCamelCase_       	=		None
      elif embedding_proj_norm_type == "layer":
        lowerCamelCase_       	=		nn.LayerNorm(__SCREAMING_SNAKE_CASE   )
      else:
        raise ValueError(F'''unsupported embedding_proj_norm_type: {embedding_proj_norm_type}'''   )

      lowerCamelCase_       	=		nn.Linear(__SCREAMING_SNAKE_CASE			,		__SCREAMING_SNAKE_CASE   )

      if encoder_hid_proj_type is None:
        lowerCamelCase_       	=		None
      elif encoder_hid_proj_type == "linear":
        lowerCamelCase_       	=		nn.Linear(__SCREAMING_SNAKE_CASE			,		__SCREAMING_SNAKE_CASE   )
      else:
        raise ValueError(F'''unsupported encoder_hid_proj_type: {encoder_hid_proj_type}'''   )

      lowerCamelCase_       	=		nn.Parameter(torch.zeros(1			,		num_embeddings + additional_embeddings			,		__SCREAMING_SNAKE_CASE   )   )

      if added_emb_type == "prd":
        lowerCamelCase_       	=		nn.Parameter(torch.zeros(1			,		1			,		__SCREAMING_SNAKE_CASE   )   )
      elif added_emb_type is None:
        lowerCamelCase_       	=		None
      else:
        raise ValueError(
            F'''`added_emb_type`: {added_emb_type} is not supported. Make sure to choose one of `\'prd\'` or `None`.'''   )

      lowerCamelCase_       	=		nn.ModuleList(
          [
              BasicTransformerBlock(
                  __SCREAMING_SNAKE_CASE			,		__SCREAMING_SNAKE_CASE			,		__SCREAMING_SNAKE_CASE			,		dropout=__SCREAMING_SNAKE_CASE			,		activation_fn='gelu'			,		attention_bias=__SCREAMING_SNAKE_CASE			,		)
              for d in range(__SCREAMING_SNAKE_CASE   )
          ]   )

      if norm_in_type == "layer":
        lowerCamelCase_       	=		nn.LayerNorm(__SCREAMING_SNAKE_CASE   )
      elif norm_in_type is None:
        lowerCamelCase_       	=		None
      else:
        raise ValueError(F'''Unsupported norm_in_type: {norm_in_type}.'''   )

      lowerCamelCase_       	=		nn.LayerNorm(__SCREAMING_SNAKE_CASE   )

      lowerCamelCase_       	=		nn.Linear(__SCREAMING_SNAKE_CASE			,		__SCREAMING_SNAKE_CASE   )

      lowerCamelCase_       	=		torch.full(
          [num_embeddings + additional_embeddings, num_embeddings + additional_embeddings]			,		-10_000.0   )
      causal_attention_mask.triu_(1   )
      lowerCamelCase_       	=		causal_attention_mask[None, ...]
      self.register_buffer('causal_attention_mask'			,		__SCREAMING_SNAKE_CASE			,		persistent=__SCREAMING_SNAKE_CASE   )

      lowerCamelCase_       	=		nn.Parameter(torch.zeros(1			,		__SCREAMING_SNAKE_CASE   )   )
      lowerCamelCase_       	=		nn.Parameter(torch.zeros(1			,		__SCREAMING_SNAKE_CASE   )   )






    @property
    # Copied from diffusers.models.unet_2d_condition.UNet2DConditionModel.attn_processors
    def   UpperCamelCase			(			self	:				List[str]   )	->						Dict[str, AttentionProcessor]:
      lowerCamelCase_       	=		{}

      def fn_recursive_add_processors(__SCREAMING_SNAKE_CASE	:				str			,		__SCREAMING_SNAKE_CASE	:				torch.nn.Module			,		__SCREAMING_SNAKE_CASE	:				Dict[str, AttentionProcessor]   ):
        if hasattr(__SCREAMING_SNAKE_CASE			,		'set_processor'   ):
          lowerCamelCase_       	=		module.processor

        for sub_name, child in module.named_children():
          fn_recursive_add_processors(F'''{name}.{sub_name}'''			,		__SCREAMING_SNAKE_CASE			,		__SCREAMING_SNAKE_CASE   )

        return processors

      for name, module in self.named_children():
        fn_recursive_add_processors(__SCREAMING_SNAKE_CASE			,		__SCREAMING_SNAKE_CASE			,		__SCREAMING_SNAKE_CASE   )

      return processors






    def   UpperCamelCase			(			self	:				Optional[Any]			,		__SCREAMING_SNAKE_CASE	:				Union[AttentionProcessor, Dict[str, AttentionProcessor]]   )	->						Optional[int]:
      lowerCamelCase_       	=		len(self.attn_processors.keys()   )

      if isinstance(__SCREAMING_SNAKE_CASE			,		__SCREAMING_SNAKE_CASE   ) and len(__SCREAMING_SNAKE_CASE   ) != count:
        raise ValueError(
            F'''A dict of processors was passed, but the number of processors {len(__SCREAMING_SNAKE_CASE   )} does not match the'''
            F''' number of attention layers: {count}. Please make sure to pass {count} processor classes.'''   )

      def fn_recursive_attn_processor(__SCREAMING_SNAKE_CASE	:				str			,		__SCREAMING_SNAKE_CASE	:				torch.nn.Module			,		__SCREAMING_SNAKE_CASE	:				str   ):
        if hasattr(__SCREAMING_SNAKE_CASE			,		'set_processor'   ):
          if not isinstance(__SCREAMING_SNAKE_CASE			,		__SCREAMING_SNAKE_CASE   ):
            module.set_processor(__SCREAMING_SNAKE_CASE   )
          else:
            module.set_processor(processor.pop(F'''{name}.processor'''   )   )

        for sub_name, child in module.named_children():
          fn_recursive_attn_processor(F'''{name}.{sub_name}'''			,		__SCREAMING_SNAKE_CASE			,		__SCREAMING_SNAKE_CASE   )

      for name, module in self.named_children():
        fn_recursive_attn_processor(__SCREAMING_SNAKE_CASE			,		__SCREAMING_SNAKE_CASE			,		__SCREAMING_SNAKE_CASE   )






    def   UpperCamelCase			(			self	:				Any   )	->						int:
      self.set_attn_processor(AttnProcessor()   )






    def   UpperCamelCase			(			self	:				str			,		__SCREAMING_SNAKE_CASE	:				Dict			,		__SCREAMING_SNAKE_CASE	:				Union[torch.Tensor, float, int]			,		__SCREAMING_SNAKE_CASE	:				torch.FloatTensor			,		__SCREAMING_SNAKE_CASE	:				Optional[torch.FloatTensor] = None			,		__SCREAMING_SNAKE_CASE	:				Optional[torch.BoolTensor] = None			,		__SCREAMING_SNAKE_CASE	:				bool = True			,		)	->						List[Any]:
      lowerCamelCase_       	=		hidden_states.shape[0]

      lowerCamelCase_       	=		timestep
      if not torch.is_tensor(__SCREAMING_SNAKE_CASE   ):
        lowerCamelCase_       	=		torch.tensor([timesteps]			,		dtype=torch.long			,		device=hidden_states.device   )
      elif torch.is_tensor(__SCREAMING_SNAKE_CASE   ) and len(timesteps.shape   ) == 0:
        lowerCamelCase_       	=		timesteps[None].to(hidden_states.device   )

      # broadcast to batch dimension in a way that's compatible with ONNX/Core ML
      lowerCamelCase_       	=		timesteps * torch.ones(__SCREAMING_SNAKE_CASE			,		dtype=timesteps.dtype			,		device=timesteps.device   )

      lowerCamelCase_       	=		self.time_proj(__SCREAMING_SNAKE_CASE   )

      # timesteps does not contain any weights and will always return f32 tensors
      # but time_embedding might be fp16, so we need to cast here.
      lowerCamelCase_       	=		timesteps_projected.to(dtype=self.dtype   )
      lowerCamelCase_       	=		self.time_embedding(__SCREAMING_SNAKE_CASE   )

      if self.embedding_proj_norm is not None:
        lowerCamelCase_       	=		self.embedding_proj_norm(__SCREAMING_SNAKE_CASE   )

      lowerCamelCase_       	=		self.embedding_proj(__SCREAMING_SNAKE_CASE   )
      if self.encoder_hidden_states_proj is not None and encoder_hidden_states is not None:
        lowerCamelCase_       	=		self.encoder_hidden_states_proj(__SCREAMING_SNAKE_CASE   )
      elif self.encoder_hidden_states_proj is not None and encoder_hidden_states is None:
        raise ValueError('`encoder_hidden_states_proj` requires `encoder_hidden_states` to be set'   )

      lowerCamelCase_       	=		self.proj_in(__SCREAMING_SNAKE_CASE   )

      lowerCamelCase_       	=		self.positional_embedding.to(hidden_states.dtype   )

      lowerCamelCase_       	=		[]
      lowerCamelCase_       	=		0

      if encoder_hidden_states is not None:
        additional_embeds.append(__SCREAMING_SNAKE_CASE   )
        additional_embeddings_len += encoder_hidden_states.shape[1]

      if len(proj_embeddings.shape   ) == 2:
        lowerCamelCase_       	=		proj_embeddings[:, None, :]

      if len(hidden_states.shape   ) == 2:
        lowerCamelCase_       	=		hidden_states[:, None, :]

      lowerCamelCase_       	=		additional_embeds + [
          proj_embeddings,
          time_embeddings[:, None, :],
          hidden_states,
      ]

      if self.prd_embedding is not None:
        lowerCamelCase_       	=		self.prd_embedding.to(hidden_states.dtype   ).expand(__SCREAMING_SNAKE_CASE			,		-1			,		-1   )
        additional_embeds.append(__SCREAMING_SNAKE_CASE   )

      lowerCamelCase_       	=		torch.cat(
          __SCREAMING_SNAKE_CASE			,		dim=1			,		)

      # Allow positional_embedding to not include the `addtional_embeddings` and instead pad it with zeros for these additional tokens
      lowerCamelCase_       	=		additional_embeddings_len + proj_embeddings.shape[1] + 1
      if positional_embeddings.shape[1] < hidden_states.shape[1]:
        lowerCamelCase_       	=		F.pad(
            __SCREAMING_SNAKE_CASE			,		(
                0,
                0,
                additional_embeddings_len,
                self.prd_embedding.shape[1] if self.prd_embedding is not None else 0,
            )			,		value=0.0			,		)

      lowerCamelCase_       	=		hidden_states + positional_embeddings

      if attention_mask is not None:
        lowerCamelCase_       	=		(1 - attention_mask.to(hidden_states.dtype   )) * -10_000.0
        lowerCamelCase_       	=		F.pad(__SCREAMING_SNAKE_CASE			,		(0, self.additional_embeddings)			,		value=0.0   )
        lowerCamelCase_       	=		(attention_mask[:, None, :] + self.causal_attention_mask).to(hidden_states.dtype   )
        lowerCamelCase_       	=		attention_mask.repeat_interleave(self.config.num_attention_heads			,		dim=0   )

      if self.norm_in is not None:
        lowerCamelCase_       	=		self.norm_in(__SCREAMING_SNAKE_CASE   )

      for block in self.transformer_blocks:
        lowerCamelCase_       	=		block(__SCREAMING_SNAKE_CASE			,		attention_mask=__SCREAMING_SNAKE_CASE   )

      lowerCamelCase_       	=		self.norm_out(__SCREAMING_SNAKE_CASE   )

      if self.prd_embedding is not None:
        lowerCamelCase_       	=		hidden_states[:, -1]
      else:
        lowerCamelCase_       	=		hidden_states[:, additional_embeddings_len:]

      lowerCamelCase_       	=		self.proj_to_clip_embeddings(__SCREAMING_SNAKE_CASE   )

      if not return_dict:
        return (predicted_image_embedding,)

      return PriorTransformerOutput(predicted_image_embedding=__SCREAMING_SNAKE_CASE   )






    def   UpperCamelCase			(			self	:				Optional[Any]			,		__SCREAMING_SNAKE_CASE	:				Optional[int]   )	->						Optional[int]:
      lowerCamelCase_       	=		(prior_latents * self.clip_std) + self.clip_mean
      return prior_latents




