
"""simple docstring"""






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      _SCREAMING_SNAKE_CASE   (     A__				):
			UpperCAmelCase_    :torch.FloatTensor




class      _SCREAMING_SNAKE_CASE   (     A__      ,					A__				):






			@register_to_config
			def __init__(   self			,  __A = None			,  __A = 3			,  __A = 3			,  __A = False			,  __A = "positional"			,  __A = 0			,  __A = True			,  __A = ("DownBlock2D", "AttnDownBlock2D", "AttnDownBlock2D", "AttnDownBlock2D")			,  __A = ("AttnUpBlock2D", "AttnUpBlock2D", "AttnUpBlock2D", "UpBlock2D")			,  __A = (224, 448, 672, 896)			,  __A = 2			,  __A = 1			,  __A = 1			,  __A = "conv"			,  __A = "conv"			,  __A = "silu"			,  __A = 8			,  __A = 32			,  __A = 1E-5			,  __A = "default"			,  __A = True			,  __A = None			,  __A = None			,  )     ->	Any:
						super().__init__()

						lowerCAmelCase_       :Dict     	=    sample_size
						lowerCAmelCase_       :int     	=    block_out_channels[0] * 4

						# Check inputs
						if len(__A			) != len(__A			):
									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(__A			) != len(__A			):
									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
						lowerCAmelCase_       :Union[str, Any]     	=    nn.Convad(__A			,  block_out_channels[0]			,  kernel_size=3			,  padding=(1, 1)			)

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

						lowerCAmelCase_       :List[Any]     	=    TimestepEmbedding(__A			,  __A			)

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

						lowerCAmelCase_       :int     	=    nn.ModuleList([]			)
						lowerCAmelCase_       :Optional[int]     	=    None
						lowerCAmelCase_       :Tuple     	=    nn.ModuleList([]			)

						# down
						lowerCAmelCase_       :Any     	=    block_out_channels[0]
						for i, down_block_type in enumerate(__A			):
									lowerCAmelCase_       :int     	=    output_channel
									lowerCAmelCase_       :Dict     	=    block_out_channels[i]
									lowerCAmelCase_       :Tuple     	=    i == len(__A			) - 1

									lowerCAmelCase_       :str     	=    get_down_block(
									    __A			,  num_layers=__A			,  in_channels=__A			,  out_channels=__A			,  temb_channels=__A			,  add_downsample=not is_final_block			,  resnet_eps=__A			,  resnet_act_fn=__A			,  resnet_groups=__A			,  attention_head_dim=attention_head_dim if attention_head_dim is not None else output_channel			,  downsample_padding=__A			,  resnet_time_scale_shift=__A			,  downsample_type=__A			,  )
									self.down_blocks.append(__A			)

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

						# up
						lowerCAmelCase_       :Optional[int]     	=    list(reversed(__A			)			)
						lowerCAmelCase_       :Optional[int]     	=    reversed_block_out_channels[0]
						for i, up_block_type in enumerate(__A			):
									lowerCAmelCase_       :str     	=    output_channel
									lowerCAmelCase_       :Dict     	=    reversed_block_out_channels[i]
									lowerCAmelCase_       :Dict     	=    reversed_block_out_channels[min(i + 1			,  len(__A			) - 1			)]

									lowerCAmelCase_       :Dict     	=    i == len(__A			) - 1

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

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






			def 					__lowerCAmelCase					(   self			,  __A			,  __A			,  __A = None			,  __A = True			,  )     ->	Union[UNetaDOutput, Tuple]:
						# 0. center input if necessary
						if self.config.center_input_sample:
									lowerCAmelCase_       :Union[str, Any]     	=    2 * sample - 1.0

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

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

						lowerCAmelCase_       :Optional[int]     	=    self.time_proj(__A			)

						# 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.
						lowerCAmelCase_       :List[str]     	=    t_emb.to(dtype=self.dtype			)
						lowerCAmelCase_       :int     	=    self.time_embedding(__A			)

						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":
												lowerCAmelCase_       :Optional[int]     	=    self.time_proj(__A			)

									lowerCAmelCase_       :int     	=    self.class_embedding(__A			).to(dtype=self.dtype			)
									lowerCAmelCase_       :List[str]     	=    emb + class_emb

						# 2. pre-process
						lowerCAmelCase_       :Tuple     	=    sample
						lowerCAmelCase_       :Optional[int]     	=    self.conv_in(__A			)

						# 3. down
						lowerCAmelCase_       :Optional[Any]     	=    (sample,)
						for downsample_block in self.down_blocks:
									if hasattr(__A			,  """skip_conv"""			):
												lowerCAmelCase_	,			lowerCAmelCase_	,			lowerCAmelCase_       :Tuple     	=    downsample_block(
												    hidden_states=__A			,  temb=__A			,  skip_sample=__A			)
									else:
												lowerCAmelCase_	,			lowerCAmelCase_       :Any     	=    downsample_block(hidden_states=__A			,  temb=__A			)

									down_block_res_samples += res_samples

						# 4. mid
						lowerCAmelCase_       :Any     	=    self.mid_block(__A			,  __A			)

						# 5. up
						lowerCAmelCase_       :Union[str, Any]     	=    None
						for upsample_block in self.up_blocks:
									lowerCAmelCase_       :Union[str, Any]     	=    down_block_res_samples[-len(upsample_block.resnets			) :]
									lowerCAmelCase_       :Any     	=    down_block_res_samples[: -len(upsample_block.resnets			)]

									if hasattr(__A			,  """skip_conv"""			):
												lowerCAmelCase_	,			lowerCAmelCase_       :Dict     	=    upsample_block(__A			,  __A			,  __A			,  __A			)
									else:
												lowerCAmelCase_       :List[str]     	=    upsample_block(__A			,  __A			,  __A			)

        # 6. post-process
						lowerCAmelCase_       :Tuple     	=    self.conv_norm_out(__A			)
						lowerCAmelCase_       :int     	=    self.conv_act(__A			)
						lowerCAmelCase_       :List[str]     	=    self.conv_out(__A			)

						if skip_sample is not None:
									sample += skip_sample

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

						if not return_dict:
									return (sample,)

						return UNetaDOutput(sample=__A			)








"""simple docstring"""






from typing import List, Optional, Tuple, Union

from . import logging
from .import_utils import is_torch_available, is_torch_version


if is_torch_available():
				import torch

__UpperCAmelCase 					=    logging.get_logger(__name__)  # pylint: disable=invalid-name

try:
				from torch._dynamo import allow_in_graph as maybe_allow_in_graph
except (ImportError, ModuleNotFoundError):






				def 		_snake_case			(      cls						:    Tuple						)    ->							List[Any]:


							'''simple docstring'''





							return cls






def 		_snake_case			(      lowercase__						:    Union[Tuple, List] ,	lowercase__						:    Optional[Union[List["torch.Generator"], "torch.Generator"]] = None ,	lowercase__						:    Optional["torch.device"] = None ,	lowercase__						:    Optional["torch.dtype"] = None ,	lowercase__						:    Optional["torch.layout"] = None ,	)    ->							Optional[Any]:


			'''simple docstring'''





			lowerCAmelCase_       :Any     	=    device
			lowerCAmelCase_       :Any     	=    shape[0]

			lowerCAmelCase_       :Dict     	=    layout or torch.strided
			lowerCAmelCase_       :List[str]     	=    device or torch.device("""cpu"""						)

			if generator is not None:
						lowerCAmelCase_       :Optional[Any]     	=    generator.device.type if not isinstance(lowercase__ ,	lowercase__						) else generator[0].device.type
						if gen_device_type != device.type and gen_device_type == "cpu":
									lowerCAmelCase_       :Optional[int]     	=    """cpu"""
									if device != "mps":
												logger.info(
												    f"""The passed generator was created on 'cpu' even though a tensor on {device} was expected."""
												    f""" Tensors will be created on 'cpu' and then moved to {device}. Note that one can probably"""
												    f""" slighly speed up this function by passing a generator that was created on the {device} device."""						)
						elif gen_device_type != device.type and gen_device_type == "cuda":
									raise ValueError(f"""Cannot generate a {device} tensor from a generator of type {gen_device_type}."""						)

			if isinstance(lowercase__ ,	lowercase__						):
						lowerCAmelCase_       :List[str]     	=    (1,) + shape[1:]
						lowerCAmelCase_       :Optional[Any]     	=    [
						    torch.randn(lowercase__ ,	generator=generator[i] ,	device=lowercase__ ,	dtype=lowercase__ ,	layout=lowercase__						)
						    for i in range(lowercase__						)
						]
						lowerCAmelCase_       :str     	=    torch.cat(lowercase__ ,	dim=0						).to(lowercase__						)
			else:
						lowerCAmelCase_       :Dict     	=    torch.randn(lowercase__ ,	generator=lowercase__ ,	device=lowercase__ ,	dtype=lowercase__ ,	layout=lowercase__						).to(lowercase__						)

			return latents






def 		_snake_case			(      lowercase__						:    Union[str, Any]						)    ->							int:


			'''simple docstring'''





			if is_torch_version("""<""" ,	"""2.0.0"""						) or not hasattr(lowercase__ ,	"""_dynamo"""						):
						return False
			return isinstance(lowercase__ ,	torch._dynamo.eval_frame.OptimizedModule						)






