


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

import torch
import torch.nn as nn

from ..configuration_utils import ConfigMixin, register_to_config
from ..utils import BaseOutput
from .embeddings import GaussianFourierProjection, TimestepEmbedding, Timesteps
from .modeling_utils import ModelMixin
from .unet_ad_blocks import UNetMidBlockaD, get_down_block, get_up_block


@dataclass
class 			__lowerCAmelCase				(					_UpperCamelCase			):


	'''simple docstring'''





	_A       	=							42


class 			__lowerCAmelCase				(					_UpperCamelCase      ,						_UpperCamelCase			):


	'''simple docstring'''





	@register_to_config
	def __init__(	self:	Tuple,				lowerCamelCase_:	Optional[Union[int, Tuple[int, int]]] = None,				lowerCamelCase_:	int = 3,				lowerCamelCase_:	int = 3,				lowerCamelCase_:	bool = False,				lowerCamelCase_:	str = "positional",				lowerCamelCase_:	int = 0,				lowerCamelCase_:	bool = True,				lowerCamelCase_:	Tuple[str] = ("DownBlock2D", "AttnDownBlock2D", "AttnDownBlock2D", "AttnDownBlock2D"),				lowerCamelCase_:	Tuple[str] = ("AttnUpBlock2D", "AttnUpBlock2D", "AttnUpBlock2D", "UpBlock2D"),				lowerCamelCase_:	Tuple[int] = (224, 448, 672, 896),				lowerCamelCase_:	int = 2,				lowerCamelCase_:	float = 1,				lowerCamelCase_:	int = 1,				lowerCamelCase_:	str = "conv",				lowerCamelCase_:	str = "conv",				lowerCamelCase_:	str = "silu",				lowerCamelCase_:	Optional[int] = 8,				lowerCamelCase_:	int = 32,				lowerCamelCase_:	float = 1E-5,				lowerCamelCase_:	str = "default",				lowerCamelCase_:	bool = True,				lowerCamelCase_:	Optional[str] = None,				lowerCamelCase_:	Optional[int] = None,				):
			super().__init__()

			lowercase__				:			Union[str, Any]            =		sample_size
			lowercase__				:			Optional[Any]            =		block_out_channels[0] * 4

			# Check inputs
			if len(lowerCamelCase_ ) != len(lowerCamelCase_ ):
					raise ValueError(
					    F"""Must provide the same number of `down_block_types` as `up_block_types`. `down_block_types`: {down_block_types}. `up_block_types`: {up_block_types}.""" )

			if len(lowerCamelCase_ ) != len(lowerCamelCase_ ):
					raise ValueError(
					    F"""Must provide the same number of `block_out_channels` as `down_block_types`. `block_out_channels`: {block_out_channels}. `down_block_types`: {down_block_types}.""" )

			# input
			lowercase__				:			Optional[int]            =		nn.Convad(lowerCamelCase_,				block_out_channels[0],				kernel_size=3,				padding=(1, 1) )

			# time
			if time_embedding_type == "fourier":
					lowercase__				:			Any            =		GaussianFourierProjection(embedding_size=block_out_channels[0],				scale=16 )
					lowercase__				:			Optional[Any]            =		2 * block_out_channels[0]
			elif time_embedding_type == "positional":
					lowercase__				:			List[Any]            =		Timesteps(block_out_channels[0],				lowerCamelCase_,				lowerCamelCase_ )
					lowercase__				:			Optional[int]            =		block_out_channels[0]

			lowercase__				:			Optional[int]            =		TimestepEmbedding(lowerCamelCase_,				lowerCamelCase_ )

			# class embedding
			if class_embed_type is None and num_class_embeds is not None:
					lowercase__				:			List[Any]            =		nn.Embedding(lowerCamelCase_,				lowerCamelCase_ )
			elif class_embed_type == "timestep":
					lowercase__				:			Optional[int]            =		TimestepEmbedding(lowerCamelCase_,				lowerCamelCase_ )
			elif class_embed_type == "identity":
					lowercase__				:			Union[str, Any]            =		nn.Identity(lowerCamelCase_,				lowerCamelCase_ )
			else:
					lowercase__				:			List[str]            =		None

			lowercase__				:			int            =		nn.ModuleList([] )
			lowercase__				:			Union[str, Any]            =		None
			lowercase__				:			Any            =		nn.ModuleList([] )

			# down
			lowercase__				:			Tuple            =		block_out_channels[0]
			for i, down_block_type in enumerate(lowerCamelCase_ ):
					lowercase__				:			Union[str, Any]            =		output_channel
					lowercase__				:			str            =		block_out_channels[i]
					lowercase__				:			Union[str, Any]            =		i == len(lowerCamelCase_ ) - 1

					lowercase__				:			Optional[Any]            =		get_down_block(
					    lowerCamelCase_,				num_layers=lowerCamelCase_,				in_channels=lowerCamelCase_,				out_channels=lowerCamelCase_,				temb_channels=lowerCamelCase_,				add_downsample=not is_final_block,				resnet_eps=lowerCamelCase_,				resnet_act_fn=lowerCamelCase_,				resnet_groups=lowerCamelCase_,				attention_head_dim=attention_head_dim if attention_head_dim is not None else output_channel,				downsample_padding=lowerCamelCase_,				resnet_time_scale_shift=lowerCamelCase_,				downsample_type=lowerCamelCase_,				)
					self.down_blocks.append(lowerCamelCase_ )

			# mid
			lowercase__				:			Dict            =		UNetMidBlockaD(
			    in_channels=block_out_channels[-1],				temb_channels=lowerCamelCase_,				resnet_eps=lowerCamelCase_,				resnet_act_fn=lowerCamelCase_,				output_scale_factor=lowerCamelCase_,				resnet_time_scale_shift=lowerCamelCase_,				attention_head_dim=attention_head_dim if attention_head_dim is not None else block_out_channels[-1],				resnet_groups=lowerCamelCase_,				add_attention=lowerCamelCase_,				)

			# up
			lowercase__				:			List[Any]            =		list(reversed(lowerCamelCase_ ) )
			lowercase__				:			str            =		reversed_block_out_channels[0]
			for i, up_block_type in enumerate(lowerCamelCase_ ):
					lowercase__				:			str            =		output_channel
					lowercase__				:			Optional[int]            =		reversed_block_out_channels[i]
					lowercase__				:			Any            =		reversed_block_out_channels[min(i + 1,				len(lowerCamelCase_ ) - 1 )]

					lowercase__				:			Dict            =		i == len(lowerCamelCase_ ) - 1

					lowercase__				:			Dict            =		get_up_block(
					    lowerCamelCase_,				num_layers=layers_per_block + 1,				in_channels=lowerCamelCase_,				out_channels=lowerCamelCase_,				prev_output_channel=lowerCamelCase_,				temb_channels=lowerCamelCase_,				add_upsample=not is_final_block,				resnet_eps=lowerCamelCase_,				resnet_act_fn=lowerCamelCase_,				resnet_groups=lowerCamelCase_,				attention_head_dim=attention_head_dim if attention_head_dim is not None else output_channel,				resnet_time_scale_shift=lowerCamelCase_,				upsample_type=lowerCamelCase_,				)
					self.up_blocks.append(lowerCamelCase_ )
					lowercase__				:			Union[str, Any]            =		output_channel

			# out
			lowercase__				:			Union[str, Any]            =		norm_num_groups if norm_num_groups is not None else min(block_out_channels[0] // 4,				32 )
			lowercase__				:			Union[str, Any]            =		nn.GroupNorm(num_channels=block_out_channels[0],				num_groups=lowerCamelCase_,				eps=lowerCamelCase_ )
			lowercase__				:			str            =		nn.SiLU()
			lowercase__				:			str            =		nn.Convad(block_out_channels[0],				lowerCamelCase_,				kernel_size=3,				padding=1 )





	def       snake_case__(	self:	List[str],				lowerCamelCase_:	torch.FloatTensor,				lowerCamelCase_:	Union[torch.Tensor, float, int],				lowerCamelCase_:	Optional[torch.Tensor] = None,				lowerCamelCase_:	bool = True,				):
			# 0. center input if necessary
			if self.config.center_input_sample:
					lowercase__				:			Any            =		2 * sample - 1.0

			# 1. time
			lowercase__				:			int            =		timestep
			if not torch.is_tensor(lowerCamelCase_ ):
					lowercase__				:			List[Any]            =		torch.tensor([timesteps],				dtype=torch.long,				device=sample.device )
			elif torch.is_tensor(lowerCamelCase_ ) and len(timesteps.shape ) == 0:
					lowercase__				:			Union[str, Any]            =		timesteps[None].to(sample.device )

			# broadcast to batch dimension in a way that's compatible with ONNX/Core ML
			lowercase__				:			str            =		timesteps * torch.ones(sample.shape[0],				dtype=timesteps.dtype,				device=timesteps.device )

			lowercase__				:			str            =		self.time_proj(lowerCamelCase_ )

			# timesteps does not contain any weights and will always return f32 tensors
			# but time_embedding might actually be running in fp16. so we need to cast here.
			# there might be better ways to encapsulate this.
			lowercase__				:			List[str]            =		t_emb.to(dtype=self.dtype )
			lowercase__				:			Any            =		self.time_embedding(lowerCamelCase_ )

			if self.class_embedding is not None:
					if class_labels is None:
							raise ValueError('class_labels should be provided when doing class conditioning' )

					if self.config.class_embed_type == "timestep":
							lowercase__				:			List[str]            =		self.time_proj(lowerCamelCase_ )

					lowercase__				:			str            =		self.class_embedding(lowerCamelCase_ ).to(dtype=self.dtype )
					lowercase__				:			str            =		emb + class_emb

			# 2. pre-process
			lowercase__				:			Dict            =		sample
			lowercase__				:			Any            =		self.conv_in(lowerCamelCase_ )

			# 3. down
			lowercase__				:			Dict            =		(sample,)
			for downsample_block in self.down_blocks:
					if hasattr(lowerCamelCase_,				'skip_conv' ):
							lowercase__				:			List[str]            =		downsample_block(
							    hidden_states=lowerCamelCase_,				temb=lowerCamelCase_,				skip_sample=lowerCamelCase_ )
					else:
							lowercase__				:			Any            =		downsample_block(hidden_states=lowerCamelCase_,				temb=lowerCamelCase_ )

					down_block_res_samples += res_samples

			# 4. mid
			lowercase__				:			Optional[int]            =		self.mid_block(lowerCamelCase_,				lowerCamelCase_ )

			# 5. up
			lowercase__				:			int            =		None
			for upsample_block in self.up_blocks:
					lowercase__				:			int            =		down_block_res_samples[-len(upsample_block.resnets ) :]
					lowercase__				:			Optional[int]            =		down_block_res_samples[: -len(upsample_block.resnets )]

					if hasattr(lowerCamelCase_,				'skip_conv' ):
							lowercase__				:			Optional[Any]            =		upsample_block(lowerCamelCase_,				lowerCamelCase_,				lowerCamelCase_,				lowerCamelCase_ )
					else:
							lowercase__				:			int            =		upsample_block(lowerCamelCase_,				lowerCamelCase_,				lowerCamelCase_ )

        # 6. post-process
			lowercase__				:			List[Any]            =		self.conv_norm_out(lowerCamelCase_ )
			lowercase__				:			Dict            =		self.conv_act(lowerCamelCase_ )
			lowercase__				:			Dict            =		self.conv_out(lowerCamelCase_ )

			if skip_sample is not None:
					sample += skip_sample

			if self.config.time_embedding_type == "fourier":
					lowercase__				:			str            =		timesteps.reshape((sample.shape[0], *([1] * len(sample.shape[1:] ))) )
					lowercase__				:			List[Any]            =		sample / timesteps

			if not return_dict:
					return (sample,)

			return UNetaDOutput(sample=lowerCamelCase_ )









from typing import List

import PIL
import torch
from PIL import Image

from ...configuration_utils import ConfigMixin
from ...models.modeling_utils import ModelMixin
from ...utils import PIL_INTERPOLATION


class 			__lowerCAmelCase				(					_UpperCamelCase      ,						_UpperCamelCase			):


	'''simple docstring'''





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

			self.register_buffer('watermark_image',				torch.zeros((62, 62, 4) ) )
			lowercase__				:			int            =		None





	def       snake_case__(	self:	Optional[int],				lowerCamelCase_:	List[PIL.Image.Image],				lowerCamelCase_:	Dict=None ):
			# copied from https://github.com/deep-floyd/IF/blob/b77482e36ca2031cb94dbca1001fc1e6400bf4ab/deepfloyd_if/modules/base.py#L287

			lowercase__				:			str            =		images[0].height
			lowercase__				:			Optional[Any]            =		images[0].width

			lowercase__				:			List[Any]            =		sample_size or h

			lowercase__				:			Any            =		min(h / sample_size,				w / sample_size )
			lowercase__      ,							lowercase__				:			Optional[int]            =		(int(h / coef ), int(w / coef )) if coef < 1 else (h, w)

			lowercase__      ,							lowercase__				:			Any            =		1024**2, img_w * img_h
			lowercase__				:			Union[str, Any]            =		(Sa / Sa) ** 0.5
			lowercase__      ,							lowercase__      ,							lowercase__				:			Optional[int]            =		int(K * 62 ), img_w - int(14 * K ), img_h - int(14 * K )

			if self.watermark_image_as_pil is None:
					lowercase__				:			Tuple            =		self.watermark_image.to(torch.uinta ).cpu().numpy()
					lowercase__				:			str            =		Image.fromarray(lowerCamelCase_,				mode='RGBA' )
					lowercase__				:			Tuple            =		watermark_image

			lowercase__				:			Optional[int]            =		self.watermark_image_as_pil.resize(
			    (wm_size, wm_size),				PIL_INTERPOLATION['bicubic'],				reducing_gap=lowerCamelCase_ )

			for pil_img in images:
					pil_img.paste(lowerCamelCase_,				box=(wm_x - wm_size, wm_y - wm_size, wm_x, wm_y),				mask=wm_img.split()[-1] )

			return images


