





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       __lowerCAmelCase					(       nn.Module      ):


	'''simple docstring'''





	def __init__(						self:						str,					lowerCamelCase_:						int,					lowerCamelCase_:						int,					lowerCamelCase_:						int,					lowerCamelCase_:						List[str]=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,					):
			super().__init__()
			lowercase__					:      Any      =		only_cross_attention

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

			# 3. Feed-forward
			lowercase__					:      Dict      =		nn.LayerNorm(lowerCamelCase_,					elementwise_affine=lowerCamelCase_			)
			lowercase__					:      int      =		FeedForward(lowerCamelCase_,					dropout=lowerCamelCase_,					activation_fn=lowerCamelCase_,					final_dropout=lowerCamelCase_			)

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

	def    snake_case__(						self:						Any,					lowerCamelCase_:						Optional[int],					lowerCamelCase_:						int			):
			# Sets chunk feed-forward
			lowercase__					:      List[str]      =		chunk_size
			lowercase__					:      Optional[Any]      =		dim





	def    snake_case__(						self:						Optional[Any],					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,					):
			# Notice that normalization is always applied before the real computation in the following blocks.
			# 1. Self-Attention
			if self.use_ada_layer_norm:
					lowercase__					:      Tuple      =		self.norma(lowerCamelCase_,					lowerCamelCase_			)
			elif self.use_ada_layer_norm_zero:
					lowercase__   ,				lowercase__   ,				lowercase__   ,				lowercase__   ,				lowercase__					:      List[Any]      =		self.norma(
					    lowerCamelCase_,					lowerCamelCase_,					lowerCamelCase_,					hidden_dtype=hidden_states.dtype			)
			else:
					lowercase__					:      Dict      =		self.norma(lowerCamelCase_			)

			lowercase__					:      Tuple      =		cross_attention_kwargs if cross_attention_kwargs is not None else {}

			lowercase__					:      List[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:
					lowercase__					:      int      =		gate_msa.unsqueeze(1			) * attn_output
			lowercase__					:      Tuple      =		attn_output + hidden_states

			# 2. Cross-Attention
			if self.attna is not None:
					lowercase__					:      Tuple      =		(
					    self.norma(lowerCamelCase_,					lowerCamelCase_			) if self.use_ada_layer_norm else self.norma(lowerCamelCase_			)
					)

					lowercase__					:      int      =		self.attna(
					    lowerCamelCase_,					encoder_hidden_states=lowerCamelCase_,					attention_mask=lowerCamelCase_,					**lowerCamelCase_,					)
					lowercase__					:      List[str]      =		attn_output + hidden_states

			# 3. Feed-forward
			lowercase__					:      Any      =		self.norma(lowerCamelCase_			)

			if self.use_ada_layer_norm_zero:
					lowercase__					:      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`."""			)

					lowercase__					:      Optional[int]      =		norm_hidden_states.shape[self._chunk_dim] // self._chunk_size
					lowercase__					:      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:
					lowercase__					:      List[str]      =		self.ff(lowerCamelCase_			)

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

			lowercase__					:      Optional[int]      =		ff_output + hidden_states

			return hidden_states


class       __lowerCAmelCase					(       nn.Module      ):


	'''simple docstring'''





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

			if activation_fn == "gelu":
					lowercase__					:      Any      =		GELU(lowerCamelCase_,					lowerCamelCase_			)
			if activation_fn == "gelu-approximate":
					lowercase__					:      List[Any]      =		GELU(lowerCamelCase_,					lowerCamelCase_,					approximate='tanh'			)
			elif activation_fn == "geglu":
					lowercase__					:      List[str]      =		GEGLU(lowerCamelCase_,					lowerCamelCase_			)
			elif activation_fn == "geglu-approximate":
					lowercase__					:      List[Any]      =		ApproximateGELU(lowerCamelCase_,					lowerCamelCase_			)

			lowercase__					:      List[str]      =		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    snake_case__(						self:						int,					lowerCamelCase_:						List[str]			):
			for module in self.net:
					lowercase__					:      Optional[Any]      =		module(lowerCamelCase_			)
			return hidden_states


class       __lowerCAmelCase					(       nn.Module      ):


	'''simple docstring'''





	def __init__(						self:						Union[str, Any],					lowerCamelCase_:						int,					lowerCamelCase_:						int,					lowerCamelCase_:						str = "none"			):
			super().__init__()
			lowercase__					:      Union[str, Any]      =		nn.Linear(lowerCamelCase_,					lowerCamelCase_			)
			lowercase__					:      List[Any]      =		approximate

	def    snake_case__(						self:						Tuple,					lowerCamelCase_:						Tuple			):
			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    snake_case__(						self:						str,					lowerCamelCase_:						Dict			):
			lowercase__					:      Union[str, Any]      =		self.proj(lowerCamelCase_			)
			lowercase__					:      Optional[int]      =		self.gelu(lowerCamelCase_			)
			return hidden_states


class       __lowerCAmelCase					(       nn.Module      ):


	'''simple docstring'''





	def __init__(						self:						List[Any],					lowerCamelCase_:						int,					lowerCamelCase_:						int			):
			super().__init__()
			lowercase__					:      Optional[int]      =		nn.Linear(lowerCamelCase_,					dim_out * 2			)

	def    snake_case__(						self:						Dict,					lowerCamelCase_:						str			):
			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    snake_case__(						self:						Optional[int],					lowerCamelCase_:						Dict			):
			lowercase__   ,				lowercase__					:      Optional[Any]      =		self.proj(lowerCamelCase_			).chunk(2,					dim=-1			)
			return hidden_states * self.gelu(lowerCamelCase_			)


class       __lowerCAmelCase					(       nn.Module      ):


	'''simple docstring'''





	def __init__(						self:						List[Any],					lowerCamelCase_:						int,					lowerCamelCase_:						int			):
			super().__init__()
			lowercase__					:      str      =		nn.Linear(lowerCamelCase_,					lowerCamelCase_			)





	def    snake_case__(						self:						List[Any],					lowerCamelCase_:						Dict			):
			lowercase__					:      str      =		self.proj(lowerCamelCase_			)
			return x * torch.sigmoid(1.7_0_2 * x			)


class       __lowerCAmelCase					(       nn.Module      ):


	'''simple docstring'''





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





	def    snake_case__(						self:						Tuple,					lowerCamelCase_:						Any,					lowerCamelCase_:						Optional[Any]			):
			lowercase__					:      Union[str, Any]      =		self.linear(self.silu(self.emb(lowerCamelCase_			)			)			)
			lowercase__   ,				lowercase__					:      Tuple      =		torch.chunk(lowerCamelCase_,					2			)
			lowercase__					:      List[str]      =		self.norm(lowerCamelCase_			) * (1 + scale) + shift
			return x


class       __lowerCAmelCase					(       nn.Module      ):


	'''simple docstring'''





	def __init__(						self:						Dict,					lowerCamelCase_:						List[str],					lowerCamelCase_:						Optional[int]			):
			super().__init__()

			lowercase__					:      List[str]      =		CombinedTimestepLabelEmbeddings(lowerCamelCase_,					lowerCamelCase_			)

			lowercase__					:      Optional[int]      =		nn.SiLU()
			lowercase__					:      Any      =		nn.Linear(lowerCamelCase_,					6 * embedding_dim,					bias=lowerCamelCase_			)
			lowercase__					:      Optional[int]      =		nn.LayerNorm(lowerCamelCase_,					elementwise_affine=lowerCamelCase_,					eps=1E-6			)





	def    snake_case__(						self:						List[str],					lowerCamelCase_:						List[str],					lowerCamelCase_:						str,					lowerCamelCase_:						str,					lowerCamelCase_:						Optional[int]=None			):
			lowercase__					:      str      =		self.linear(self.silu(self.emb(lowerCamelCase_,					lowerCamelCase_,					hidden_dtype=lowerCamelCase_			)			)			)
			lowercase__   ,				lowercase__   ,				lowercase__   ,				lowercase__   ,				lowercase__   ,				lowercase__					:      int      =		emb.chunk(6,					dim=1			)
			lowercase__					:      int      =		self.norm(lowerCamelCase_			) * (1 + scale_msa[:, None]) + shift_msa[:, None]
			return x, gate_msa, shift_mlp, scale_mlp, gate_mlp


class       __lowerCAmelCase					(       nn.Module      ):


	'''simple docstring'''





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

			if act_fn is None:
					lowercase__					:      Any      =		None
			else:
					lowercase__					:      int      =		get_activation(lowerCamelCase_			)

			lowercase__					:      Optional[Any]      =		nn.Linear(lowerCamelCase_,					out_dim * 2			)





	def    snake_case__(						self:						Dict,					lowerCamelCase_:						Optional[Any],					lowerCamelCase_:						List[Any]			):
			if self.act:
					lowercase__					:      Union[str, Any]      =		self.act(lowerCamelCase_			)
			lowercase__					:      Any      =		self.linear(lowerCamelCase_			)
			lowercase__					:      Union[str, Any]      =		emb[:, :, None, None]
			lowercase__   ,				lowercase__					:      Tuple      =		emb.chunk(2,					dim=1			)

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









import numpy as np
from cva import destroyAllWindows, imread, imshow, waitKey


class       __lowerCAmelCase					:


	'''simple docstring'''





	def __init__(						self:						List[str],					lowerCamelCase_:						Optional[Any],					lowerCamelCase_:						int,					lowerCamelCase_:						int			):
			if dst_width < 0 or dst_height < 0:
					raise ValueError('Destination width/height should be > 0'			)

			lowercase__					:      str      =		img
			lowercase__					:      Tuple      =		img.shape[1]
			lowercase__					:      str      =		img.shape[0]
			lowercase__					:      Tuple      =		dst_width
			lowercase__					:      Any      =		dst_height

			lowercase__					:      Tuple      =		self.src_w / self.dst_w
			lowercase__					:      Union[str, Any]      =		self.src_h / self.dst_h

			lowercase__					:      List[Any]      =		(
			    np.ones((self.dst_h, self.dst_w, 3),					np.uinta			) * 255
			)

	def    snake_case__(						self:						Dict			):
			for i in range(self.dst_h			):
					for j in range(self.dst_w			):
							lowercase__					:      Tuple      =		self.img[self.get_y(lowerCamelCase_			)][self.get_x(lowerCamelCase_			)]

	def    snake_case__(						self:						List[str],					lowerCamelCase_:						int			):
			return int(self.ratio_x * x			)





	def    snake_case__(						self:						int,					lowerCamelCase_:						int			):
			return int(self.ratio_y * y			)


if __name__ == "__main__":
			__UpperCamelCase,	__UpperCamelCase:	List[Any] = 8_0_0, 6_0_0
			__UpperCamelCase:	Optional[Any] = imread("""image_data/lena.jpg""", 1)
			__UpperCamelCase:	Union[str, Any] = NearestNeighbour(im, dst_w, dst_h)
			n.process()

			imshow(
			    F'Image resized from: {im.shape[1]}x{im.shape[0]} to {dst_w}x{dst_h}', n.output
			)
			waitKey(0)
			destroyAllWindows()


