
import math
from typing import Optional

import numpy as np
import torch
from torch import nn

from .activations import get_activation




def 		lowerCAmelCase_     (     _SCREAMING_SNAKE_CASE      ,	_SCREAMING_SNAKE_CASE      ,	_SCREAMING_SNAKE_CASE = False      ,	_SCREAMING_SNAKE_CASE = 1      ,	_SCREAMING_SNAKE_CASE = 1      ,	_SCREAMING_SNAKE_CASE = 1_0_0_0_0      ,	)   ->      int:

							"""simple docstring"""






							assert len(timesteps.shape  ) == 1, "Timesteps should be a 1d-array"

							_UpperCAmelCase  :			Union[str, Any]	       =		embedding_dim // 2
							_UpperCAmelCase  :			Optional[Any]	       =		-math.log(_SCREAMING_SNAKE_CASE  ) * torch.arange(
							    start=0      ,	end=_SCREAMING_SNAKE_CASE      ,	dtype=torch.floataa      ,	device=timesteps.device  )
							_UpperCAmelCase  :			Optional[int]	       =		exponent / (half_dim - downscale_freq_shift)

							_UpperCAmelCase  :			Dict	       =		torch.exp(_SCREAMING_SNAKE_CASE  )
							_UpperCAmelCase  :			Tuple	       =		timesteps[:, None].float() * emb[None, :]

							# scale embeddings
							_UpperCAmelCase  :			Optional[int]	       =		scale * emb

							# concat sine and cosine embeddings
							_UpperCAmelCase  :			List[str]	       =		torch.cat([torch.sin(_SCREAMING_SNAKE_CASE  ), torch.cos(_SCREAMING_SNAKE_CASE  )]      ,	dim=-1  )

							# flip sine and cosine embeddings
							if flip_sin_to_cos:
														_UpperCAmelCase  :			Optional[Any]	       =		torch.cat([emb[:, half_dim:], emb[:, :half_dim]]      ,	dim=-1  )

							# zero pad
							if embedding_dim % 2 == 1:
														_UpperCAmelCase  :			Dict	       =		torch.nn.functional.pad(_SCREAMING_SNAKE_CASE      ,	(0, 1, 0, 0)  )
							return emb




def 		lowerCAmelCase_     (     _SCREAMING_SNAKE_CASE      ,	_SCREAMING_SNAKE_CASE      ,	_SCREAMING_SNAKE_CASE=False      ,	_SCREAMING_SNAKE_CASE=0  )   ->      Union[str, Any]:

							"""simple docstring"""






							_UpperCAmelCase  :			List[Any]	       =		np.arange(_SCREAMING_SNAKE_CASE      ,	dtype=np.floataa  )
							_UpperCAmelCase  :			Dict	       =		np.arange(_SCREAMING_SNAKE_CASE      ,	dtype=np.floataa  )
							_UpperCAmelCase  :			List[Any]	       =		np.meshgrid(_SCREAMING_SNAKE_CASE      ,	_SCREAMING_SNAKE_CASE  )  # here w goes first
							_UpperCAmelCase  :			int	       =		np.stack(_SCREAMING_SNAKE_CASE      ,	axis=0  )

							_UpperCAmelCase  :			Optional[int]	       =		grid.reshape([2, 1, grid_size, grid_size]  )
							_UpperCAmelCase  :			List[Any]	       =		get_ad_sincos_pos_embed_from_grid(_SCREAMING_SNAKE_CASE      ,	_SCREAMING_SNAKE_CASE  )
							if cls_token and extra_tokens > 0:
														_UpperCAmelCase  :			List[str]	       =		np.concatenate([np.zeros([extra_tokens, embed_dim]  ), pos_embed]      ,	axis=0  )
							return pos_embed




def 		lowerCAmelCase_     (     _SCREAMING_SNAKE_CASE      ,	_SCREAMING_SNAKE_CASE  )   ->      Any:

							"""simple docstring"""






							if embed_dim % 2 != 0:
														raise ValueError("embed_dim must be divisible by 2"  )

							# use half of dimensions to encode grid_h
							_UpperCAmelCase  :			Union[str, Any]	       =		get_ad_sincos_pos_embed_from_grid(embed_dim // 2      ,	grid[0]  )  # (H*W, D/2)
							_UpperCAmelCase  :			Optional[Any]	       =		get_ad_sincos_pos_embed_from_grid(embed_dim // 2      ,	grid[1]  )  # (H*W, D/2)

							_UpperCAmelCase  :			Union[str, Any]	       =		np.concatenate([emb_h, emb_w]      ,	axis=1  )  # (H*W, D)
							return emb




def 		lowerCAmelCase_     (     _SCREAMING_SNAKE_CASE      ,	_SCREAMING_SNAKE_CASE  )   ->      Optional[int]:

							"""simple docstring"""






							if embed_dim % 2 != 0:
														raise ValueError("embed_dim must be divisible by 2"  )

							_UpperCAmelCase  :			Tuple	       =		np.arange(embed_dim // 2      ,	dtype=np.floataa  )
							omega /= embed_dim / 2.0
							_UpperCAmelCase  :			Optional[int]	       =		1.0 / 1_0_0_0_0**omega  # (D/2,)

							_UpperCAmelCase  :			Any	       =		pos.reshape(-1  )  # (M,)
							_UpperCAmelCase  :			int	       =		np.einsum("m,d->md"      ,	_SCREAMING_SNAKE_CASE      ,	_SCREAMING_SNAKE_CASE  )  # (M, D/2), outer product

							_UpperCAmelCase  :			str	       =		np.sin(_SCREAMING_SNAKE_CASE  )  # (M, D/2)
							_UpperCAmelCase  :			Optional[Any]	       =		np.cos(_SCREAMING_SNAKE_CASE  )  # (M, D/2)

							_UpperCAmelCase  :			List[Any]	       =		np.concatenate([emb_sin, emb_cos]      ,	axis=1  )  # (M, D)
							return emb
class    _UpperCamelCase(   nn.Module				):

						def __init__(					self		:			Any	,						_lowerCamelCase		:			Union[str, Any]=2_24	,						_lowerCamelCase		:			Tuple=2_24	,						_lowerCamelCase		:			Optional[int]=16	,						_lowerCamelCase		:			Optional[int]=3	,						_lowerCamelCase		:			Tuple=7_68	,						_lowerCamelCase		:			int=False	,						_lowerCamelCase		:			List[Any]=True	,						_lowerCamelCase		:			Optional[Any]=True	,						):
													super().__init__()

													_UpperCAmelCase  :			List[str]	       =		(height // patch_size) * (width // patch_size)
													_UpperCAmelCase  :			Union[str, Any]	       =		flatten
													_UpperCAmelCase  :			Any	       =		layer_norm

													_UpperCAmelCase  :			Any	       =		nn.Convad(
													    _lowerCamelCase	,						_lowerCamelCase	,						kernel_size=(patch_size, patch_size)	,						stride=_lowerCamelCase	,						bias=_lowerCamelCase			)
													if layer_norm:
																				_UpperCAmelCase  :			List[str]	       =		nn.LayerNorm(_lowerCamelCase	,						elementwise_affine=_lowerCamelCase	,						eps=1E-6			)
													else:
																				_UpperCAmelCase  :			str	       =		None

													_UpperCAmelCase  :			Dict	       =		get_ad_sincos_pos_embed(_lowerCamelCase	,						int(num_patches**0.5			)			)
													self.register_buffer("pos_embed"	,						torch.from_numpy(_lowerCamelCase			).float().unsqueeze(0			)	,						persistent=_lowerCamelCase			)






						def 						a__ (					self		:			Tuple	,						_lowerCamelCase		:			Union[str, Any]			):
													_UpperCAmelCase  :			Optional[int]	       =		self.proj(_lowerCamelCase			)
													if self.flatten:
																				_UpperCAmelCase  :			Optional[Any]	       =		latent.flatten(2			).transpose(1	,						2			)  # BCHW -> BNC
													if self.layer_norm:
																				_UpperCAmelCase  :			Dict	       =		self.norm(_lowerCamelCase			)
													return latent + self.pos_embed







class    _UpperCamelCase(   nn.Module				):

						def __init__(					self		:			Optional[Any]	,						_lowerCamelCase		:			int	,						_lowerCamelCase		:			int	,						_lowerCamelCase		:			str = "silu"	,						_lowerCamelCase		:			int = None	,						_lowerCamelCase		:			Optional[str] = None	,						_lowerCamelCase		:			Any=None	,						):
													super().__init__()

													_UpperCAmelCase  :			Optional[int]	       =		nn.Linear(_lowerCamelCase	,						_lowerCamelCase			)

													if cond_proj_dim is not None:
																				_UpperCAmelCase  :			Tuple	       =		nn.Linear(_lowerCamelCase	,						_lowerCamelCase	,						bias=_lowerCamelCase			)
													else:
																				_UpperCAmelCase  :			Any	       =		None

													_UpperCAmelCase  :			int	       =		get_activation(_lowerCamelCase			)

													if out_dim is not None:
																				_UpperCAmelCase  :			Optional[int]	       =		out_dim
													else:
																				_UpperCAmelCase  :			Any	       =		time_embed_dim
													_UpperCAmelCase  :			List[Any]	       =		nn.Linear(_lowerCamelCase	,						_lowerCamelCase			)

													if post_act_fn is None:
																				_UpperCAmelCase  :			List[Any]	       =		None
													else:
																				_UpperCAmelCase  :			Tuple	       =		get_activation(_lowerCamelCase			)






						def 						a__ (					self		:			Optional[int]	,						_lowerCamelCase		:			Tuple	,						_lowerCamelCase		:			List[str]=None			):
													if condition is not None:
																				_UpperCAmelCase  :			int	       =		sample + self.cond_proj(_lowerCamelCase			)
													_UpperCAmelCase  :			Dict	       =		self.linear_a(_lowerCamelCase			)

													if self.act is not None:
																				_UpperCAmelCase  :			Union[str, Any]	       =		self.act(_lowerCamelCase			)

													_UpperCAmelCase  :			Tuple	       =		self.linear_a(_lowerCamelCase			)

													if self.post_act is not None:
																				_UpperCAmelCase  :			Optional[int]	       =		self.post_act(_lowerCamelCase			)
													return sample







class    _UpperCamelCase(   nn.Module				):

						def __init__(					self		:			Tuple	,						_lowerCamelCase		:			int	,						_lowerCamelCase		:			bool	,						_lowerCamelCase		:			float			):
													super().__init__()
													_UpperCAmelCase  :			int	       =		num_channels
													_UpperCAmelCase  :			List[Any]	       =		flip_sin_to_cos
													_UpperCAmelCase  :			str	       =		downscale_freq_shift






						def 						a__ (					self		:			Union[str, Any]	,						_lowerCamelCase		:			List[Any]			):
													_UpperCAmelCase  :			Optional[Any]	       =		get_timestep_embedding(
													    _lowerCamelCase	,						self.num_channels	,						flip_sin_to_cos=self.flip_sin_to_cos	,						downscale_freq_shift=self.downscale_freq_shift	,						)
													return t_emb







class    _UpperCamelCase(   nn.Module				):

						def __init__(					self		:			Any	,						_lowerCamelCase		:			int = 2_56	,						_lowerCamelCase		:			float = 1.0	,						_lowerCamelCase		:			Any=True	,						_lowerCamelCase		:			Any=True	,						_lowerCamelCase		:			Optional[Any]=False			):
													super().__init__()
													_UpperCAmelCase  :			Tuple	       =		nn.Parameter(torch.randn(_lowerCamelCase			) * scale	,						requires_grad=_lowerCamelCase			)
													_UpperCAmelCase  :			int	       =		log
													_UpperCAmelCase  :			List[Any]	       =		flip_sin_to_cos

													if set_W_to_weight:
																				# to delete later
																				_UpperCAmelCase  :			Optional[int]	       =		nn.Parameter(torch.randn(_lowerCamelCase			) * scale	,						requires_grad=_lowerCamelCase			)

																				_UpperCAmelCase  :			List[str]	       =		self.W






						def 						a__ (					self		:			Union[str, Any]	,						_lowerCamelCase		:			Tuple			):
													if self.log:
																				_UpperCAmelCase  :			List[str]	       =		torch.log(_lowerCamelCase			)

													_UpperCAmelCase  :			List[Any]	       =		x[:, None] * self.weight[None, :] * 2 * np.pi

													if self.flip_sin_to_cos:
																				_UpperCAmelCase  :			Optional[int]	       =		torch.cat([torch.cos(_lowerCamelCase			), torch.sin(_lowerCamelCase			)]	,						dim=-1			)
													else:
																				_UpperCAmelCase  :			Optional[int]	       =		torch.cat([torch.sin(_lowerCamelCase			), torch.cos(_lowerCamelCase			)]	,						dim=-1			)
													return out







class    _UpperCamelCase(   nn.Module				):

						def __init__(					self		:			Union[str, Any]	,						_lowerCamelCase		:			int	,						_lowerCamelCase		:			int	,						_lowerCamelCase		:			int	,						_lowerCamelCase		:			int	,						):
													super().__init__()

													_UpperCAmelCase  :			List[str]	       =		height
													_UpperCAmelCase  :			List[str]	       =		width
													_UpperCAmelCase  :			Union[str, Any]	       =		num_embed
													_UpperCAmelCase  :			Optional[int]	       =		embed_dim

													_UpperCAmelCase  :			Dict	       =		nn.Embedding(self.num_embed	,						_lowerCamelCase			)
													_UpperCAmelCase  :			Optional[int]	       =		nn.Embedding(self.height	,						_lowerCamelCase			)
													_UpperCAmelCase  :			Union[str, Any]	       =		nn.Embedding(self.width	,						_lowerCamelCase			)






						def 						a__ (					self		:			Optional[int]	,						_lowerCamelCase		:			Dict			):
													_UpperCAmelCase  :			List[Any]	       =		self.emb(_lowerCamelCase			)

													_UpperCAmelCase  :			Tuple	       =		self.height_emb(torch.arange(self.height	,						device=index.device			).view(1	,						self.height			)			)

													# 1 x H x D -> 1 x H x 1 x D
													_UpperCAmelCase  :			List[str]	       =		height_emb.unsqueeze(2			)

													_UpperCAmelCase  :			Tuple	       =		self.width_emb(torch.arange(self.width	,						device=index.device			).view(1	,						self.width			)			)

													# 1 x W x D -> 1 x 1 x W x D
													_UpperCAmelCase  :			Optional[Any]	       =		width_emb.unsqueeze(1			)

													_UpperCAmelCase  :			int	       =		height_emb + width_emb

													# 1 x H x W x D -> 1 x L xD
													_UpperCAmelCase  :			Tuple	       =		pos_emb.view(1	,						self.height * self.width	,						-1			)

													_UpperCAmelCase  :			List[str]	       =		emb + pos_emb[:, : emb.shape[1], :]

													return emb







class    _UpperCamelCase(   nn.Module				):

						def __init__(					self		:			Optional[int]	,						_lowerCamelCase		:			Any	,						_lowerCamelCase		:			int	,						_lowerCamelCase		:			Optional[Any]			):
													super().__init__()
													_UpperCAmelCase  :			str	       =		dropout_prob > 0
													_UpperCAmelCase  :			Dict	       =		nn.Embedding(num_classes + use_cfg_embedding	,						_lowerCamelCase			)
													_UpperCAmelCase  :			Dict	       =		num_classes
													_UpperCAmelCase  :			Union[str, Any]	       =		dropout_prob

						def 						a__ (					self		:			Tuple	,						_lowerCamelCase		:			Union[str, Any]	,						_lowerCamelCase		:			Dict=None			):
													if force_drop_ids is None:
																				_UpperCAmelCase  :			List[Any]	       =		torch.rand(labels.shape[0]	,						device=labels.device			) < self.dropout_prob
													else:
																				_UpperCAmelCase  :			Optional[Any]	       =		torch.tensor(force_drop_ids == 1			)
													_UpperCAmelCase  :			Any	       =		torch.where(_lowerCamelCase	,						self.num_classes	,						_lowerCamelCase			)
													return labels






						def 						a__ (					self		:			List[str]	,						_lowerCamelCase		:			torch.LongTensor	,						_lowerCamelCase		:			Optional[Any]=None			):
													_UpperCAmelCase  :			Optional[Any]	       =		self.dropout_prob > 0
													if (self.training and use_dropout) or (force_drop_ids is not None):
																				_UpperCAmelCase  :			Dict	       =		self.token_drop(_lowerCamelCase	,						_lowerCamelCase			)
													_UpperCAmelCase  :			Optional[int]	       =		self.embedding_table(_lowerCamelCase			)
													return embeddings







class    _UpperCamelCase(   nn.Module				):

						def __init__(					self		:			Tuple	,						_lowerCamelCase		:			int = 10_24	,						_lowerCamelCase		:			int = 7_68	,						_lowerCamelCase		:			int = 7_68	,						_lowerCamelCase		:			int = 10	,						):
													super().__init__()

													_UpperCAmelCase  :			Tuple	       =		num_image_text_embeds
													_UpperCAmelCase  :			Union[str, Any]	       =		nn.Linear(_lowerCamelCase	,						self.num_image_text_embeds * cross_attention_dim			)
													_UpperCAmelCase  :			int	       =		nn.Linear(_lowerCamelCase	,						_lowerCamelCase			)






						def 						a__ (					self		:			List[str]	,						_lowerCamelCase		:			torch.FloatTensor	,						_lowerCamelCase		:			torch.FloatTensor			):
													_UpperCAmelCase  :			Optional[int]	       =		text_embeds.shape[0]

													# image
													_UpperCAmelCase  :			str	       =		self.image_embeds(_lowerCamelCase			)
													_UpperCAmelCase  :			Union[str, Any]	       =		image_text_embeds.reshape(_lowerCamelCase	,						self.num_image_text_embeds	,						-1			)

													# text
													_UpperCAmelCase  :			Tuple	       =		self.text_proj(_lowerCamelCase			)

													return torch.cat([image_text_embeds, text_embeds]	,						dim=1			)







class    _UpperCamelCase(   nn.Module				):

						def __init__(					self		:			Tuple	,						_lowerCamelCase		:			int = 7_68	,						_lowerCamelCase		:			int = 7_68	,						_lowerCamelCase		:			int = 32	,						):
													super().__init__()

													_UpperCAmelCase  :			List[Any]	       =		num_image_text_embeds
													_UpperCAmelCase  :			Any	       =		nn.Linear(_lowerCamelCase	,						self.num_image_text_embeds * cross_attention_dim			)
													_UpperCAmelCase  :			Optional[int]	       =		nn.LayerNorm(_lowerCamelCase			)






						def 						a__ (					self		:			Any	,						_lowerCamelCase		:			torch.FloatTensor			):
													_UpperCAmelCase  :			Optional[Any]	       =		image_embeds.shape[0]

													# image
													_UpperCAmelCase  :			Union[str, Any]	       =		self.image_embeds(_lowerCamelCase			)
													_UpperCAmelCase  :			Dict	       =		image_embeds.reshape(_lowerCamelCase	,						self.num_image_text_embeds	,						-1			)
													_UpperCAmelCase  :			Optional[int]	       =		self.norm(_lowerCamelCase			)
													return image_embeds







class    _UpperCamelCase(   nn.Module				):

						def __init__(					self		:			Tuple	,						_lowerCamelCase		:			Union[str, Any]	,						_lowerCamelCase		:			Union[str, Any]	,						_lowerCamelCase		:			Optional[int]=0.1			):
													super().__init__()

													_UpperCAmelCase  :			List[Any]	       =		Timesteps(num_channels=2_56	,						flip_sin_to_cos=_lowerCamelCase	,						downscale_freq_shift=1			)
													_UpperCAmelCase  :			List[str]	       =		TimestepEmbedding(in_channels=2_56	,						time_embed_dim=_lowerCamelCase			)
													_UpperCAmelCase  :			Optional[Any]	       =		LabelEmbedding(_lowerCamelCase	,						_lowerCamelCase	,						_lowerCamelCase			)






						def 						a__ (					self		:			List[Any]	,						_lowerCamelCase		:			str	,						_lowerCamelCase		:			Optional[int]	,						_lowerCamelCase		:			str=None			):
													_UpperCAmelCase  :			List[Any]	       =		self.time_proj(_lowerCamelCase			)
													_UpperCAmelCase  :			Any	       =		self.timestep_embedder(timesteps_proj.to(dtype=_lowerCamelCase			)			)  # (N, D)

													_UpperCAmelCase  :			Dict	       =		self.class_embedder(_lowerCamelCase			)  # (N, D)

													_UpperCAmelCase  :			Tuple	       =		timesteps_emb + class_labels  # (N, D)

													return conditioning







class    _UpperCamelCase(   nn.Module				):

						def __init__(					self		:			List[str]	,						_lowerCamelCase		:			int	,						_lowerCamelCase		:			int	,						_lowerCamelCase		:			int = 64			):
													super().__init__()
													_UpperCAmelCase  :			str	       =		nn.LayerNorm(_lowerCamelCase			)
													_UpperCAmelCase  :			Optional[int]	       =		AttentionPooling(_lowerCamelCase	,						_lowerCamelCase			)
													_UpperCAmelCase  :			Union[str, Any]	       =		nn.Linear(_lowerCamelCase	,						_lowerCamelCase			)
													_UpperCAmelCase  :			Optional[Any]	       =		nn.LayerNorm(_lowerCamelCase			)






						def 						a__ (					self		:			Optional[Any]	,						_lowerCamelCase		:			Union[str, Any]			):
													_UpperCAmelCase  :			List[str]	       =		self.norma(_lowerCamelCase			)
													_UpperCAmelCase  :			List[str]	       =		self.pool(_lowerCamelCase			)
													_UpperCAmelCase  :			List[str]	       =		self.proj(_lowerCamelCase			)
													_UpperCAmelCase  :			Optional[int]	       =		self.norma(_lowerCamelCase			)
													return hidden_states







class    _UpperCamelCase(   nn.Module				):

						def __init__(					self		:			List[str]	,						_lowerCamelCase		:			int = 7_68	,						_lowerCamelCase		:			int = 7_68	,						_lowerCamelCase		:			int = 15_36			):
													super().__init__()
													_UpperCAmelCase  :			List[str]	       =		nn.Linear(_lowerCamelCase	,						_lowerCamelCase			)
													_UpperCAmelCase  :			Any	       =		nn.LayerNorm(_lowerCamelCase			)
													_UpperCAmelCase  :			List[str]	       =		nn.Linear(_lowerCamelCase	,						_lowerCamelCase			)






						def 						a__ (					self		:			Optional[int]	,						_lowerCamelCase		:			torch.FloatTensor	,						_lowerCamelCase		:			torch.FloatTensor			):
													# text
													_UpperCAmelCase  :			str	       =		self.text_proj(_lowerCamelCase			)
													_UpperCAmelCase  :			str	       =		self.text_norm(_lowerCamelCase			)

													# image
													_UpperCAmelCase  :			Optional[Any]	       =		self.image_proj(_lowerCamelCase			)

													return time_image_embeds + time_text_embeds







class    _UpperCamelCase(   nn.Module				):

						def __init__(					self		:			List[Any]	,						_lowerCamelCase		:			int = 7_68	,						_lowerCamelCase		:			int = 15_36			):
													super().__init__()
													_UpperCAmelCase  :			Optional[Any]	       =		nn.Linear(_lowerCamelCase	,						_lowerCamelCase			)
													_UpperCAmelCase  :			int	       =		nn.LayerNorm(_lowerCamelCase			)






						def 						a__ (					self		:			Optional[Any]	,						_lowerCamelCase		:			torch.FloatTensor			):
													# image
													_UpperCAmelCase  :			str	       =		self.image_proj(_lowerCamelCase			)
													_UpperCAmelCase  :			Optional[int]	       =		self.image_norm(_lowerCamelCase			)
													return time_image_embeds







class    _UpperCamelCase(   nn.Module				):

						def __init__(					self		:			Any	,						_lowerCamelCase		:			int = 7_68	,						_lowerCamelCase		:			int = 15_36			):
													super().__init__()
													_UpperCAmelCase  :			Any	       =		nn.Linear(_lowerCamelCase	,						_lowerCamelCase			)
													_UpperCAmelCase  :			List[Any]	       =		nn.LayerNorm(_lowerCamelCase			)
													_UpperCAmelCase  :			List[str]	       =		nn.Sequential(
													    nn.Convad(3	,						16	,						3	,						padding=1			)	,						nn.SiLU()	,						nn.Convad(16	,						16	,						3	,						padding=1			)	,						nn.SiLU()	,						nn.Convad(16	,						32	,						3	,						padding=1	,						stride=2			)	,						nn.SiLU()	,						nn.Convad(32	,						32	,						3	,						padding=1			)	,						nn.SiLU()	,						nn.Convad(32	,						96	,						3	,						padding=1	,						stride=2			)	,						nn.SiLU()	,						nn.Convad(96	,						96	,						3	,						padding=1			)	,						nn.SiLU()	,						nn.Convad(96	,						2_56	,						3	,						padding=1	,						stride=2			)	,						nn.SiLU()	,						nn.Convad(2_56	,						4	,						3	,						padding=1			)	,						)






						def 						a__ (					self		:			List[str]	,						_lowerCamelCase		:			torch.FloatTensor	,						_lowerCamelCase		:			torch.FloatTensor			):
													# image
													_UpperCAmelCase  :			List[Any]	       =		self.image_proj(_lowerCamelCase			)
													_UpperCAmelCase  :			List[Any]	       =		self.image_norm(_lowerCamelCase			)
													_UpperCAmelCase  :			Optional[int]	       =		self.input_hint_block(_lowerCamelCase			)
													return time_image_embeds, hint







class    _UpperCamelCase(   nn.Module				):

						def __init__(					self		:			Tuple	,						_lowerCamelCase		:			List[Any]	,						_lowerCamelCase		:			Any	,						_lowerCamelCase		:			Union[str, Any]=None			):
													super().__init__()
													_UpperCAmelCase  :			Union[str, Any]	       =		dtype
													_UpperCAmelCase  :			Union[str, Any]	       =		nn.Parameter(torch.randn(1	,						_lowerCamelCase			) / embed_dim**0.5			)
													_UpperCAmelCase  :			Optional[int]	       =		nn.Linear(_lowerCamelCase	,						_lowerCamelCase	,						dtype=self.dtype			)
													_UpperCAmelCase  :			Any	       =		nn.Linear(_lowerCamelCase	,						_lowerCamelCase	,						dtype=self.dtype			)
													_UpperCAmelCase  :			List[Any]	       =		nn.Linear(_lowerCamelCase	,						_lowerCamelCase	,						dtype=self.dtype			)
													_UpperCAmelCase  :			Optional[int]	       =		num_heads
													_UpperCAmelCase  :			Optional[Any]	       =		embed_dim // self.num_heads






						def 						a__ (					self		:			Optional[int]	,						_lowerCamelCase		:			Dict			):
													_UpperCAmelCase       ,_UpperCAmelCase       ,_UpperCAmelCase  :			Any	       =		x.size()

													def shape(_lowerCamelCase		:			List[Any]			):
																				# (bs, length, width) --> (bs, length, n_heads, dim_per_head)
																				_UpperCAmelCase  :			List[str]	       =		x.view(_lowerCamelCase	,						-1	,						self.num_heads	,						self.dim_per_head			)
																				# (bs, length, n_heads, dim_per_head) --> (bs, n_heads, length, dim_per_head)
																				_UpperCAmelCase  :			Dict	       =		x.transpose(1	,						2			)
																				# (bs, n_heads, length, dim_per_head) --> (bs*n_heads, length, dim_per_head)
																				_UpperCAmelCase  :			List[str]	       =		x.reshape(bs * self.num_heads	,						-1	,						self.dim_per_head			)
																				# (bs*n_heads, length, dim_per_head) --> (bs*n_heads, dim_per_head, length)
																				_UpperCAmelCase  :			int	       =		x.transpose(1	,						2			)
																				return x

													_UpperCAmelCase  :			int	       =		x.mean(dim=1	,						keepdim=_lowerCamelCase			) + self.positional_embedding.to(x.dtype			)
													_UpperCAmelCase  :			int	       =		torch.cat([class_token, x]	,						dim=1			)  # (bs, length+1, width)

													# (bs*n_heads, class_token_length, dim_per_head)
													_UpperCAmelCase  :			str	       =		shape(self.q_proj(_lowerCamelCase			)			)
													# (bs*n_heads, length+class_token_length, dim_per_head)
													_UpperCAmelCase  :			Dict	       =		shape(self.k_proj(_lowerCamelCase			)			)
													_UpperCAmelCase  :			Dict	       =		shape(self.v_proj(_lowerCamelCase			)			)

													# (bs*n_heads, class_token_length, length+class_token_length):
													_UpperCAmelCase  :			int	       =		1 / math.sqrt(math.sqrt(self.dim_per_head			)			)
													_UpperCAmelCase  :			Dict	       =		torch.einsum("bct,bcs->bts"	,						q * scale	,						k * scale			)  # More stable with f16 than dividing afterwards
													_UpperCAmelCase  :			Optional[int]	       =		torch.softmax(weight.float()	,						dim=-1			).type(weight.dtype			)

													# (bs*n_heads, dim_per_head, class_token_length)
													_UpperCAmelCase  :			str	       =		torch.einsum("bts,bcs->bct"	,						_lowerCamelCase	,						_lowerCamelCase			)

													# (bs, length+1, width)
													_UpperCAmelCase  :			str	       =		a.reshape(_lowerCamelCase	,						-1	,						1			).transpose(1	,						2			)

													return a[:, 0, :]  # cls_token






import gc
import random
import unittest

import numpy as np
import torch
from transformers import XLMRobertaTokenizerFast

from diffusers import DDIMScheduler, KandinskyPipeline, KandinskyPriorPipeline, UNetaDConditionModel, VQModel
from diffusers.pipelines.kandinsky.text_encoder import MCLIPConfig, MultilingualCLIP
from diffusers.utils import floats_tensor, load_numpy, slow, torch_device
from diffusers.utils.testing_utils import enable_full_determinism, require_torch_gpu

from ..test_pipelines_common import PipelineTesterMixin, assert_mean_pixel_difference


enable_full_determinism()
class    _UpperCamelCase(   SCREAMING_SNAKE_CASE		,						unittest.TestCase				):
						__A:						Dict				   =      KandinskyPipeline
						__A:						List[Any]				   =      [
						    """prompt""",
						    """image_embeds""",
						    """negative_image_embeds""",
						]
						__A:						Dict				   =      ["""prompt""", """negative_prompt""", """image_embeds""", """negative_image_embeds"""]
						__A:						List[str]				   =      [
						    """generator""",
						    """height""",
						    """width""",
						    """latents""",
						    """guidance_scale""",
						    """negative_prompt""",
						    """num_inference_steps""",
						    """return_dict""",
						    """guidance_scale""",
						    """num_images_per_prompt""",
						    """output_type""",
						    """return_dict""",
						]
						__A:						List[str]				   =      False

						@property
						def 						a__ (					self		:			List[Any]			):
													return 32

						@property
						def 						a__ (					self		:			Tuple			):
													return 32

						@property
						def 						a__ (					self		:			List[Any]			):
													return self.time_input_dim

						@property
						def 						a__ (					self		:			List[Any]			):
													return self.time_input_dim * 4

						@property
						def 						a__ (					self		:			Dict			):
													return 1_00

						@property
						def 						a__ (					self		:			Tuple			):
													_UpperCAmelCase  :			Optional[int]	       =		XLMRobertaTokenizerFast.from_pretrained("YiYiXu/tiny-random-mclip-base"			)
													return tokenizer

						@property
						def 						a__ (					self		:			Tuple			):
													torch.manual_seed(0			)
													_UpperCAmelCase  :			int	       =		MCLIPConfig(
													    numDims=self.cross_attention_dim	,						transformerDimensions=self.text_embedder_hidden_size	,						hidden_size=self.text_embedder_hidden_size	,						intermediate_size=37	,						num_attention_heads=4	,						num_hidden_layers=5	,						vocab_size=10_05	,						)

													_UpperCAmelCase  :			Optional[Any]	       =		MultilingualCLIP(_lowerCamelCase			)
													_UpperCAmelCase  :			str	       =		text_encoder.eval()

													return text_encoder

						@property
						def 						a__ (					self		:			Any			):
													torch.manual_seed(0			)

													_UpperCAmelCase  :			Any	       =		{
													    "in_channels": 4,
													    # Out channels is double in channels because predicts mean and variance
													    "out_channels": 8,
													    "addition_embed_type": "text_image",
													    "down_block_types": ("ResnetDownsampleBlock2D", "SimpleCrossAttnDownBlock2D"),
													    "up_block_types": ("SimpleCrossAttnUpBlock2D", "ResnetUpsampleBlock2D"),
													    "mid_block_type": "UNetMidBlock2DSimpleCrossAttn",
													    "block_out_channels": (self.block_out_channels_a, self.block_out_channels_a * 2),
													    "layers_per_block": 1,
													    "encoder_hid_dim": self.text_embedder_hidden_size,
													    "encoder_hid_dim_type": "text_image_proj",
													    "cross_attention_dim": self.cross_attention_dim,
													    "attention_head_dim": 4,
													    "resnet_time_scale_shift": "scale_shift",
													    "class_embed_type": None,
													}

													_UpperCAmelCase  :			Optional[Any]	       =		UNetaDConditionModel(**_lowerCamelCase			)
													return model

						@property
						def 						a__ (					self		:			Optional[Any]			):
													return {
													    "block_out_channels": [32, 64],
													    "down_block_types": ["DownEncoderBlock2D", "AttnDownEncoderBlock2D"],
													    "in_channels": 3,
													    "latent_channels": 4,
													    "layers_per_block": 1,
													    "norm_num_groups": 8,
													    "norm_type": "spatial",
													    "num_vq_embeddings": 12,
													    "out_channels": 3,
													    "up_block_types": [
													        "AttnUpDecoderBlock2D",
													        "UpDecoderBlock2D",
													    ],
													    "vq_embed_dim": 4,
													}

						@property
						def 						a__ (					self		:			Tuple			):
													torch.manual_seed(0			)
													_UpperCAmelCase  :			Dict	       =		VQModel(**self.dummy_movq_kwargs			)
													return model

						def 						a__ (					self		:			List[Any]			):
													_UpperCAmelCase  :			Union[str, Any]	       =		self.dummy_text_encoder
													_UpperCAmelCase  :			Any	       =		self.dummy_tokenizer
													_UpperCAmelCase  :			List[Any]	       =		self.dummy_unet
													_UpperCAmelCase  :			Union[str, Any]	       =		self.dummy_movq

													_UpperCAmelCase  :			str	       =		DDIMScheduler(
													    num_train_timesteps=10_00	,						beta_schedule="linear"	,						beta_start=0.0_00_85	,						beta_end=0.0_12	,						clip_sample=_lowerCamelCase	,						set_alpha_to_one=_lowerCamelCase	,						steps_offset=1	,						prediction_type="epsilon"	,						thresholding=_lowerCamelCase	,						)

													_UpperCAmelCase  :			Optional[Any]	       =		{
													    "text_encoder": text_encoder,
													    "tokenizer": tokenizer,
													    "unet": unet,
													    "scheduler": scheduler,
													    "movq": movq,
													}
													return components

						def 						a__ (					self		:			int	,						_lowerCamelCase		:			Any	,						_lowerCamelCase		:			List[str]=0			):
													_UpperCAmelCase  :			Tuple	       =		floats_tensor((1, self.cross_attention_dim)	,						rng=random.Random(_lowerCamelCase			)			).to(_lowerCamelCase			)
													_UpperCAmelCase  :			Tuple	       =		floats_tensor((1, self.cross_attention_dim)	,						rng=random.Random(seed + 1			)			).to(_lowerCamelCase			)
													if str(_lowerCamelCase			).startswith("mps"			):
																				_UpperCAmelCase  :			Union[str, Any]	       =		torch.manual_seed(_lowerCamelCase			)
													else:
																				_UpperCAmelCase  :			Optional[Any]	       =		torch.Generator(device=_lowerCamelCase			).manual_seed(_lowerCamelCase			)
													_UpperCAmelCase  :			int	       =		{
													    "prompt": "horse",
													    "image_embeds": image_embeds,
													    "negative_image_embeds": negative_image_embeds,
													    "generator": generator,
													    "height": 64,
													    "width": 64,
													    "guidance_scale": 4.0,
													    "num_inference_steps": 2,
													    "output_type": "np",
													}
													return inputs






						def 						a__ (					self		:			List[Any]			):
													_UpperCAmelCase  :			Optional[Any]	       =		"cpu"

													_UpperCAmelCase  :			Dict	       =		self.get_dummy_components()

													_UpperCAmelCase  :			Any	       =		self.pipeline_class(**_lowerCamelCase			)
													_UpperCAmelCase  :			Optional[int]	       =		pipe.to(_lowerCamelCase			)

													pipe.set_progress_bar_config(disable=_lowerCamelCase			)

													_UpperCAmelCase  :			Dict	       =		pipe(**self.get_dummy_inputs(_lowerCamelCase			)			)
													_UpperCAmelCase  :			int	       =		output.images

													_UpperCAmelCase  :			Tuple	       =		pipe(
													    **self.get_dummy_inputs(_lowerCamelCase			)	,						return_dict=_lowerCamelCase	,						)[0]

													_UpperCAmelCase  :			Any	       =		image[0, -3:, -3:, -1]
													_UpperCAmelCase  :			Optional[Any]	       =		image_from_tuple[0, -3:, -3:, -1]

													assert image.shape == (1, 64, 64, 3)

													_UpperCAmelCase  :			Dict	       =		np.array(
													    [0.32_86_63, 1.0, 0.23_21_68_73, 1.0, 0.92_71_75_64, 0.4_63_90_46, 0.96_89_47_77, 0.31_71_33_78, 0.6_29_39_53]			)

													assert (
													    np.abs(image_slice.flatten() - expected_slice			).max() < 1E-2
													), f""" expected_slice {expected_slice}, but got {image_slice.flatten()}"""
													assert (
													    np.abs(image_from_tuple_slice.flatten() - expected_slice			).max() < 1E-2
													), f""" expected_slice {expected_slice}, but got {image_from_tuple_slice.flatten()}"""







@slow
@require_torch_gpu
class    _UpperCamelCase(   unittest.TestCase				):

						def 						a__ (					self		:			int			):
													# clean up the VRAM after each test
													super().tearDown()
													gc.collect()
													torch.cuda.empty_cache()






						def 						a__ (					self		:			str			):
													_UpperCAmelCase  :			List[str]	       =		load_numpy(
													    "https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main"
													    "/kandinsky/kandinsky_text2img_cat_fp16.npy"			)

													_UpperCAmelCase  :			Dict	       =		KandinskyPriorPipeline.from_pretrained(
													    "kandinsky-community/kandinsky-2-1-prior"	,						torch_dtype=torch.floataa			)
													pipe_prior.to(_lowerCamelCase			)

													_UpperCAmelCase  :			str	       =		KandinskyPipeline.from_pretrained("kandinsky-community/kandinsky-2-1"	,						torch_dtype=torch.floataa			)
													_UpperCAmelCase  :			Union[str, Any]	       =		pipeline.to(_lowerCamelCase			)
													pipeline.set_progress_bar_config(disable=_lowerCamelCase			)

													_UpperCAmelCase  :			Tuple	       =		"red cat, 4k photo"

													_UpperCAmelCase  :			List[Any]	       =		torch.Generator(device="cuda"			).manual_seed(0			)
													_UpperCAmelCase       ,_UpperCAmelCase  :			int	       =		pipe_prior(
													    _lowerCamelCase	,						generator=_lowerCamelCase	,						num_inference_steps=5	,						negative_prompt=""	,						).to_tuple()

													_UpperCAmelCase  :			Tuple	       =		torch.Generator(device="cuda"			).manual_seed(0			)
													_UpperCAmelCase  :			Dict	       =		pipeline(
													    _lowerCamelCase	,						image_embeds=_lowerCamelCase	,						negative_image_embeds=_lowerCamelCase	,						generator=_lowerCamelCase	,						num_inference_steps=1_00	,						output_type="np"	,						)

													_UpperCAmelCase  :			Tuple	       =		output.images[0]

													assert image.shape == (5_12, 5_12, 3)

													assert_mean_pixel_difference(_lowerCamelCase	,						_lowerCamelCase			)




