"""simple docstring"""






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

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

from ..configuration_utils import ConfigMixin, register_to_config
from ..models.embeddings import ImagePositionalEmbeddings
from ..utils import BaseOutput, deprecate
from .attention import BasicTransformerBlock
from .embeddings import PatchEmbed
from .modeling_utils import ModelMixin
@dataclass
class   UpperCAmelCase__				(    a__			):


	"""simple docstring"""
	lowerCAmelCase__    :					torch.FloatTensor




class   UpperCAmelCase__				(    a__,      a__			):


	"""simple docstring"""







	@register_to_config
	def __init__(			self       ,     _SCREAMING_SNAKE_CASE = 1_6       ,     _SCREAMING_SNAKE_CASE = 8_8       ,     _SCREAMING_SNAKE_CASE = None       ,     _SCREAMING_SNAKE_CASE = None       ,     _SCREAMING_SNAKE_CASE = 1       ,     _SCREAMING_SNAKE_CASE = 0.0       ,     _SCREAMING_SNAKE_CASE = 3_2       ,     _SCREAMING_SNAKE_CASE = None       ,     _SCREAMING_SNAKE_CASE = False       ,     _SCREAMING_SNAKE_CASE = None       ,     _SCREAMING_SNAKE_CASE = None       ,     _SCREAMING_SNAKE_CASE = None       ,     _SCREAMING_SNAKE_CASE = "geglu"       ,     _SCREAMING_SNAKE_CASE = None       ,     _SCREAMING_SNAKE_CASE = False       ,     _SCREAMING_SNAKE_CASE = False       ,     _SCREAMING_SNAKE_CASE = False       ,     _SCREAMING_SNAKE_CASE = "layer_norm"       ,     _SCREAMING_SNAKE_CASE = True       ,     )    ->	Union[str, Any]:
							super().__init__()
							a_							:			Tuple     		=					use_linear_projection
							a_							:			int     		=					num_attention_heads
							a_							:			Dict     		=					attention_head_dim
							a_							:			Optional[int]     		=					num_attention_heads * attention_head_dim

							# 1. Transformer2DModel can process both standard continuous images of shape `(batch_size, num_channels, width, height)` as well as quantized image embeddings of shape `(batch_size, num_image_vectors)`
							# Define whether input is continuous or discrete depending on configuration
							a_							:			Optional[Any]     		=					(in_channels is not None) and (patch_size is None)
							a_							:			Tuple     		=					num_vector_embeds is not None
							a_							:			Optional[int]     		=					in_channels is not None and patch_size is not None

							if norm_type == "layer_norm" and num_embeds_ada_norm is not None:
													a_							:			Any     		=					(
													    f'''The configuration file of this model: {self.__class__} is outdated. `norm_type` is either not set or'''
													    ' incorrectly set to `\'layer_norm\'`.Make sure to set `norm_type` to `\'ada_norm\'` in the config.'
													    ' Please make sure to update the config accordingly as leaving `norm_type` might led to incorrect'
													    ' results in future versions. If you have downloaded this checkpoint from the Hugging Face Hub, it'
													    ' would be very nice if you could open a Pull request for the `transformer/config.json` file'
													)
													deprecate("norm_type!=num_embeds_ada_norm"       ,     "1.0.0"       ,     SCREAMING_SNAKE_CASE_       ,     standard_warn=SCREAMING_SNAKE_CASE_ )
													a_							:			Any     		=					'ada_norm'

							if self.is_input_continuous and self.is_input_vectorized:
													raise ValueError(
													    f'''Cannot define both `in_channels`: {in_channels} and `num_vector_embeds`: {num_vector_embeds}. Make'''
													    " sure that either `in_channels` or `num_vector_embeds` is None." )
							elif self.is_input_vectorized and self.is_input_patches:
													raise ValueError(
													    f'''Cannot define both `num_vector_embeds`: {num_vector_embeds} and `patch_size`: {patch_size}. Make'''
													    " sure that either `num_vector_embeds` or `num_patches` is None." )
							elif not self.is_input_continuous and not self.is_input_vectorized and not self.is_input_patches:
													raise ValueError(
													    f'''Has to define `in_channels`: {in_channels}, `num_vector_embeds`: {num_vector_embeds}, or patch_size:'''
													    f''' {patch_size}. Make sure that `in_channels`, `num_vector_embeds` or `num_patches` is not None.''' )

							# 2. Define input layers
							if self.is_input_continuous:
													a_							:			Optional[int]     		=					in_channels

													a_							:			str     		=					torch.nn.GroupNorm(num_groups=SCREAMING_SNAKE_CASE_       ,     num_channels=SCREAMING_SNAKE_CASE_       ,     eps=1E-6       ,     affine=SCREAMING_SNAKE_CASE_ )
													if use_linear_projection:
																			a_							:			Dict     		=					nn.Linear(SCREAMING_SNAKE_CASE_       ,     SCREAMING_SNAKE_CASE_ )
													else:
																			a_							:			Any     		=					nn.Convad(SCREAMING_SNAKE_CASE_       ,     SCREAMING_SNAKE_CASE_       ,     kernel_size=1       ,     stride=1       ,     padding=0 )
							elif self.is_input_vectorized:
													assert sample_size is not None, "Transformer2DModel over discrete input must provide sample_size"
													assert num_vector_embeds is not None, "Transformer2DModel over discrete input must provide num_embed"

													a_							:			Union[str, Any]     		=					sample_size
													a_							:			int     		=					sample_size
													a_							:			Optional[Any]     		=					num_vector_embeds
													a_							:			str     		=					self.height * self.width

													a_							:			Union[str, Any]     		=					ImagePositionalEmbeddings(
													    num_embed=SCREAMING_SNAKE_CASE_       ,     embed_dim=SCREAMING_SNAKE_CASE_       ,     height=self.height       ,     width=self.width )
							elif self.is_input_patches:
													assert sample_size is not None, "Transformer2DModel over patched input must provide sample_size"

													a_							:			int     		=					sample_size
													a_							:			Optional[Any]     		=					sample_size

													a_							:			List[str]     		=					patch_size
													a_							:			Tuple     		=					PatchEmbed(
													    height=SCREAMING_SNAKE_CASE_       ,     width=SCREAMING_SNAKE_CASE_       ,     patch_size=SCREAMING_SNAKE_CASE_       ,     in_channels=SCREAMING_SNAKE_CASE_       ,     embed_dim=SCREAMING_SNAKE_CASE_       ,     )

							# 3. Define transformers blocks
							a_							:			Any     		=					nn.ModuleList(
							    [
							        BasicTransformerBlock(
							            SCREAMING_SNAKE_CASE_       ,     SCREAMING_SNAKE_CASE_       ,     SCREAMING_SNAKE_CASE_       ,     dropout=SCREAMING_SNAKE_CASE_       ,     cross_attention_dim=SCREAMING_SNAKE_CASE_       ,     activation_fn=SCREAMING_SNAKE_CASE_       ,     num_embeds_ada_norm=SCREAMING_SNAKE_CASE_       ,     attention_bias=SCREAMING_SNAKE_CASE_       ,     only_cross_attention=SCREAMING_SNAKE_CASE_       ,     upcast_attention=SCREAMING_SNAKE_CASE_       ,     norm_type=SCREAMING_SNAKE_CASE_       ,     norm_elementwise_affine=SCREAMING_SNAKE_CASE_       ,     )
							        for d in range(SCREAMING_SNAKE_CASE_ )
							    ] )

							# 4. Define output layers
							a_							:			Optional[int]     		=					in_channels if out_channels is None else out_channels
							if self.is_input_continuous:
													# TODO: should use out_channels for continuous projections
													if use_linear_projection:
																			a_							:			Optional[Any]     		=					nn.Linear(SCREAMING_SNAKE_CASE_       ,     SCREAMING_SNAKE_CASE_ )
													else:
																			a_							:			List[str]     		=					nn.Convad(SCREAMING_SNAKE_CASE_       ,     SCREAMING_SNAKE_CASE_       ,     kernel_size=1       ,     stride=1       ,     padding=0 )
							elif self.is_input_vectorized:
													a_							:			Optional[Any]     		=					nn.LayerNorm(SCREAMING_SNAKE_CASE_ )
													a_							:			Any     		=					nn.Linear(SCREAMING_SNAKE_CASE_       ,     self.num_vector_embeds - 1 )
							elif self.is_input_patches:
													a_							:			List[str]     		=					nn.LayerNorm(SCREAMING_SNAKE_CASE_       ,     elementwise_affine=SCREAMING_SNAKE_CASE_       ,     eps=1E-6 )
													a_							:			Tuple     		=					nn.Linear(SCREAMING_SNAKE_CASE_       ,     2 * inner_dim )
													a_							:			Any     		=					nn.Linear(SCREAMING_SNAKE_CASE_       ,     patch_size * patch_size * self.out_channels )







	def       A    (			self       ,     _SCREAMING_SNAKE_CASE       ,     _SCREAMING_SNAKE_CASE = None       ,     _SCREAMING_SNAKE_CASE = None       ,     _SCREAMING_SNAKE_CASE = None       ,     _SCREAMING_SNAKE_CASE = None       ,     _SCREAMING_SNAKE_CASE = None       ,     _SCREAMING_SNAKE_CASE = None       ,     _SCREAMING_SNAKE_CASE = True       ,     )    ->	Optional[Any]:
							# ensure attention_mask is a bias, and give it a singleton query_tokens dimension.
							#   we may have done this conversion already, e.g. if we came here via UNet2DConditionModel#forward.
							#   we can tell by counting dims; if ndim == 2: it's a mask rather than a bias.
							# expects mask of shape:
							#   [batch, key_tokens]
							# adds singleton query_tokens dimension:
							#   [batch,                    1, key_tokens]
							# this helps to broadcast it as a bias over attention scores, which will be in one of the following shapes:
							#   [batch,  heads, query_tokens, key_tokens] (e.g. torch sdp attn)
							#   [batch * heads, query_tokens, key_tokens] (e.g. xformers or classic attn)
							if attention_mask is not None and attention_mask.ndim == 2:
													# assume that mask is expressed as:
													#   (1 = keep,      0 = discard)
													# convert mask into a bias that can be added to attention scores:
													#       (keep = +0,     discard = -10000.0)
													a_							:			List[str]     		=					(1 - attention_mask.to(hidden_states.dtype )) * -1_0_0_0_0.0
													a_							:			Tuple     		=					attention_mask.unsqueeze(1 )

							# convert encoder_attention_mask to a bias the same way we do for attention_mask
							if encoder_attention_mask is not None and encoder_attention_mask.ndim == 2:
													a_							:			int     		=					(1 - encoder_attention_mask.to(hidden_states.dtype )) * -1_0_0_0_0.0
													a_							:			Dict     		=					encoder_attention_mask.unsqueeze(1 )

							# 1. Input
							if self.is_input_continuous:
													a_							:			Union[str, Any]     		=					hidden_states.shape
													a_							:			str     		=					hidden_states

													a_							:			Union[str, Any]     		=					self.norm(SCREAMING_SNAKE_CASE_ )
													if not self.use_linear_projection:
																			a_							:			Tuple     		=					self.proj_in(SCREAMING_SNAKE_CASE_ )
																			a_							:			Optional[int]     		=					hidden_states.shape[1]
																			a_							:			Tuple     		=					hidden_states.permute(0       ,     2       ,     3       ,     1 ).reshape(SCREAMING_SNAKE_CASE_       ,     height * width       ,     SCREAMING_SNAKE_CASE_ )
													else:
																			a_							:			Optional[int]     		=					hidden_states.shape[1]
																			a_							:			Dict     		=					hidden_states.permute(0       ,     2       ,     3       ,     1 ).reshape(SCREAMING_SNAKE_CASE_       ,     height * width       ,     SCREAMING_SNAKE_CASE_ )
																			a_							:			Union[str, Any]     		=					self.proj_in(SCREAMING_SNAKE_CASE_ )
							elif self.is_input_vectorized:
													a_							:			Union[str, Any]     		=					self.latent_image_embedding(SCREAMING_SNAKE_CASE_ )
							elif self.is_input_patches:
													a_							:			Any     		=					self.pos_embed(SCREAMING_SNAKE_CASE_ )

							# 2. Blocks
							for block in self.transformer_blocks:
													a_							:			Optional[int]     		=					block(
													    SCREAMING_SNAKE_CASE_       ,     attention_mask=SCREAMING_SNAKE_CASE_       ,     encoder_hidden_states=SCREAMING_SNAKE_CASE_       ,     encoder_attention_mask=SCREAMING_SNAKE_CASE_       ,     timestep=SCREAMING_SNAKE_CASE_       ,     cross_attention_kwargs=SCREAMING_SNAKE_CASE_       ,     class_labels=SCREAMING_SNAKE_CASE_       ,     )

							# 3. Output
							if self.is_input_continuous:
													if not self.use_linear_projection:
																			a_							:			Tuple     		=					hidden_states.reshape(SCREAMING_SNAKE_CASE_       ,     SCREAMING_SNAKE_CASE_       ,     SCREAMING_SNAKE_CASE_       ,     SCREAMING_SNAKE_CASE_ ).permute(0       ,     3       ,     1       ,     2 ).contiguous()
																			a_							:			int     		=					self.proj_out(SCREAMING_SNAKE_CASE_ )
													else:
																			a_							:			Optional[int]     		=					self.proj_out(SCREAMING_SNAKE_CASE_ )
																			a_							:			Optional[int]     		=					hidden_states.reshape(SCREAMING_SNAKE_CASE_       ,     SCREAMING_SNAKE_CASE_       ,     SCREAMING_SNAKE_CASE_       ,     SCREAMING_SNAKE_CASE_ ).permute(0       ,     3       ,     1       ,     2 ).contiguous()

													a_							:			Any     		=					hidden_states + residual
							elif self.is_input_vectorized:
													a_							:			List[Any]     		=					self.norm_out(SCREAMING_SNAKE_CASE_ )
													a_							:			Optional[int]     		=					self.out(SCREAMING_SNAKE_CASE_ )
													# (batch, self.num_vector_embeds - 1, self.num_latent_pixels)
													a_							:			Tuple     		=					logits.permute(0       ,     2       ,     1 )

													# log(p(x_0))
													a_							:			List[str]     		=					F.log_softmax(logits.double()       ,     dim=1 ).float()
							elif self.is_input_patches:
													# TODO: cleanup!
													a_							:			int     		=					self.transformer_blocks[0].norma.emb(
													    SCREAMING_SNAKE_CASE_       ,     SCREAMING_SNAKE_CASE_       ,     hidden_dtype=hidden_states.dtype )
													a_							:			int     		=					self.proj_out_a(F.silu(SCREAMING_SNAKE_CASE_ ) ).chunk(2       ,     dim=1 )
													a_							:			Dict     		=					self.norm_out(SCREAMING_SNAKE_CASE_ ) * (1 + scale[:, None]) + shift[:, None]
													a_							:			Any     		=					self.proj_out_a(SCREAMING_SNAKE_CASE_ )

													# unpatchify
													a_							:			Union[str, Any]     		=					int(hidden_states.shape[1] ** 0.5 )
													a_							:			Tuple     		=					hidden_states.reshape(
													    shape=(-1, height, width, self.patch_size, self.patch_size, self.out_channels) )
													a_							:			int     		=					torch.einsum("nhwpqc->nchpwq"       ,     SCREAMING_SNAKE_CASE_ )
													a_							:			Any     		=					hidden_states.reshape(
													    shape=(-1, self.out_channels, height * self.patch_size, width * self.patch_size) )

							if not return_dict:
													return (output,)

							return TransformeraDModelOutput(sample=SCREAMING_SNAKE_CASE_ )





"""simple docstring"""






import unittest

import torch

from diffusers import DDIMScheduler, TextToVideoZeroPipeline
from diffusers.utils import load_pt, require_torch_gpu, slow

from ..test_pipelines_common import assert_mean_pixel_difference
@slow
@require_torch_gpu
class   UpperCAmelCase__				(    unittest.TestCase			):


	"""simple docstring"""







	def       A    (			self )    ->	List[str]:
							a_							:			List[str]     		=					"runwayml/stable-diffusion-v1-5"
							a_							:			Optional[int]     		=					TextToVideoZeroPipeline.from_pretrained(_SCREAMING_SNAKE_CASE       ,     torch_dtype=torch.floataa ).to("cuda" )
							a_							:			Dict     		=					DDIMScheduler.from_config(pipe.scheduler.config )
							a_							:			Optional[Any]     		=					torch.Generator(device="cuda" ).manual_seed(0 )

							a_							:			List[Any]     		=					"A bear is playing a guitar on Times Square"
							a_							:			List[Any]     		=					pipe(prompt=_SCREAMING_SNAKE_CASE       ,     generator=_SCREAMING_SNAKE_CASE ).images

							a_							:			Optional[int]     		=					load_pt(
							    "https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/text-to-video/A bear is playing a guitar on Times Square.pt" )

							assert_mean_pixel_difference(_SCREAMING_SNAKE_CASE       ,     _SCREAMING_SNAKE_CASE )




