




'''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      UpperCamelCase_  (a__  ):







						"""simple docstring"""


						_lowerCAmelCase         =       42






class      UpperCamelCase_  (a__,   a__  ):







						"""simple docstring"""


						@register_to_config
						def __init__(  self   :							Optional[Any]				,			_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				,			):





											"""simple docstring"""







											super().__init__()

											A_			:  Dict			=				sample_size
											A_			:  int			=				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
											A_			:  Tuple			=				nn.Convad(_lowerCamelCase				,			block_out_channels[0]				,			kernel_size=3				,			padding=(1, 1)     )

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

											A_			:  str			=				TimestepEmbedding(_lowerCamelCase				,			_lowerCamelCase     )

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

											A_			:  Union[str, Any]			=				nn.ModuleList([]     )
											A_			:  Tuple			=				None
											A_			:  Tuple			=				nn.ModuleList([]     )

											# down
											A_			:  Dict			=				block_out_channels[0]
											for i, down_block_type in enumerate(_lowerCamelCase     ):
																A_			:  List[Any]			=				output_channel
																A_			:  List[str]			=				block_out_channels[i]
																A_			:  List[str]			=				i == len(_lowerCamelCase     ) - 1

																A_			:  Union[str, 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
											A_			:  Tuple			=				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
											A_			:  Any			=				list(reversed(_lowerCamelCase     )     )
											A_			:  Dict			=				reversed_block_out_channels[0]
											for i, up_block_type in enumerate(_lowerCamelCase     ):
																A_			:  Optional[Any]			=				output_channel
																A_			:  Optional[Any]			=				reversed_block_out_channels[i]
																A_			:  List[Any]			=				reversed_block_out_channels[min(i + 1				,			len(_lowerCamelCase     ) - 1     )]

																A_			:  Optional[int]			=				i == len(_lowerCamelCase     ) - 1

																A_			:  List[Any]			=				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     )
																A_			:  Optional[Any]			=				output_channel

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







						def    _a     (  self   :							str				,			_lowerCamelCase   :							torch.FloatTensor				,			_lowerCamelCase   :							Union[torch.Tensor, float, int]				,			_lowerCamelCase   :							Optional[torch.Tensor] = None				,			_lowerCamelCase   :							bool = True				,			):





											"""simple docstring"""







											if self.config.center_input_sample:
																A_			:  Tuple			=				2 * sample - 1.0

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

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

											A_			:  Optional[int]			=				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.
											A_			:  List[Any]			=				t_emb.to(dtype=self.dtype     )
											A_			:  List[str]			=				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":
																					A_			:  Tuple			=				self.time_proj(_lowerCamelCase     )

																A_			:  Union[str, Any]			=				self.class_embedding(_lowerCamelCase     ).to(dtype=self.dtype     )
																A_			:  List[Any]			=				emb + class_emb

											# 2. pre-process
											A_			:  List[str]			=				sample
											A_			:  Union[str, Any]			=				self.conv_in(_lowerCamelCase     )

											# 3. down
											A_			:  Tuple			=				(sample,)
											for downsample_block in self.down_blocks:
																if hasattr(_lowerCamelCase				,			'''skip_conv'''     ):
																					A_			,A_			,A_			:  Tuple			=				downsample_block(
																					    hidden_states=_lowerCamelCase				,			temb=_lowerCamelCase				,			skip_sample=_lowerCamelCase     )
																else:
																					A_			,A_			:  Dict			=				downsample_block(hidden_states=_lowerCamelCase				,			temb=_lowerCamelCase     )

																down_block_res_samples += res_samples

											# 4. mid
											A_			:  Optional[int]			=				self.mid_block(_lowerCamelCase				,			_lowerCamelCase     )

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

																if hasattr(_lowerCamelCase				,			'''skip_conv'''     ):
																					A_			,A_			:  Tuple			=				upsample_block(_lowerCamelCase				,			_lowerCamelCase				,			_lowerCamelCase				,			_lowerCamelCase     )
																else:
																					A_			:  str			=				upsample_block(_lowerCamelCase				,			_lowerCamelCase				,			_lowerCamelCase     )

        # 6. post-process
											A_			:  Optional[int]			=				self.conv_norm_out(_lowerCamelCase     )
											A_			:  Any			=				self.conv_act(_lowerCamelCase     )
											A_			:  Union[str, Any]			=				self.conv_out(_lowerCamelCase     )

											if skip_sample is not None:
																sample += skip_sample

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

											if not return_dict:
																return (sample,)

											return UNetaDOutput(sample=_lowerCamelCase     )









'''simple docstring'''




import abc
import copy
import dataclasses
from dataclasses import dataclass
from typing import ClassVar, Dict, Type, TypeVar

from ..features import Features


snake_case__ 							=    TypeVar("""T""", bound="""TaskTemplate""")






@dataclass(frozen=a__  )
class      UpperCamelCase_  (abc.ABC  ):







						"""simple docstring"""


						_lowerCAmelCase         =       42
						_lowerCAmelCase         =       42
						_lowerCAmelCase         =       42







						def    _a     (  self   :							T				,			_lowerCamelCase   :							Features     ):





											"""simple docstring"""







											return copy.deepcopy(self     )







						@property
						def    _a     (  self   :							List[str]     ):





											"""simple docstring"""







											return Features(**self.input_schema				,			**self.label_schema     )







						@property
						@abc.abstractmethod
						def    _a     (  self   :							int     ):





											"""simple docstring"""







											raise NotImplementedError







						@classmethod
						def    _a     (  cls   :							Type[T]				,			_lowerCamelCase   :							dict     ):





											"""simple docstring"""







											A_			:  Dict			=				{f.name for f in dataclasses.fields(cls     )}
											return cls(**{k: v for k, v in template_dict.items() if k in field_names}     )



