


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 	__a   (       UpperCAmelCase    ):
		_a			:						torch.FloatTensor



class 	__a   (       UpperCAmelCase					,   UpperCAmelCase    ):
		@register_to_config
		def __init__(    self    ,					_SCREAMING_SNAKE_CASE = None    ,					_SCREAMING_SNAKE_CASE = 3    ,					_SCREAMING_SNAKE_CASE = 3    ,					_SCREAMING_SNAKE_CASE = False    ,					_SCREAMING_SNAKE_CASE = "positional"    ,					_SCREAMING_SNAKE_CASE = 0    ,					_SCREAMING_SNAKE_CASE = True    ,					_SCREAMING_SNAKE_CASE = ("DownBlock2D", "AttnDownBlock2D", "AttnDownBlock2D", "AttnDownBlock2D")    ,					_SCREAMING_SNAKE_CASE = ("AttnUpBlock2D", "AttnUpBlock2D", "AttnUpBlock2D", "UpBlock2D")    ,					_SCREAMING_SNAKE_CASE = (224, 448, 672, 896)    ,					_SCREAMING_SNAKE_CASE = 2    ,					_SCREAMING_SNAKE_CASE = 1    ,					_SCREAMING_SNAKE_CASE = 1    ,					_SCREAMING_SNAKE_CASE = "conv"    ,					_SCREAMING_SNAKE_CASE = "conv"    ,					_SCREAMING_SNAKE_CASE = "silu"    ,					_SCREAMING_SNAKE_CASE = 8    ,					_SCREAMING_SNAKE_CASE = 32    ,					_SCREAMING_SNAKE_CASE = 1e-5    ,					_SCREAMING_SNAKE_CASE = "default"    ,					_SCREAMING_SNAKE_CASE = True    ,					_SCREAMING_SNAKE_CASE = None    ,					_SCREAMING_SNAKE_CASE = None    ,					)				-> Dict:




									"""simple docstring"""





									super().__init__()

									_UpperCAmelCase					=    sample_size
									_UpperCAmelCase					=    block_out_channels[0] * 4

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

									# time
									if time_embedding_type == "fourier":
																_UpperCAmelCase					=    GaussianFourierProjection(embedding_size=block_out_channels[0]    ,					scale=16		)
																_UpperCAmelCase					=    2 * block_out_channels[0]
									elif time_embedding_type == "positional":
																_UpperCAmelCase					=    Timesteps(block_out_channels[0]    ,					_SCREAMING_SNAKE_CASE    ,					_SCREAMING_SNAKE_CASE		)
																_UpperCAmelCase					=    block_out_channels[0]

									_UpperCAmelCase					=    TimestepEmbedding(_SCREAMING_SNAKE_CASE    ,					_SCREAMING_SNAKE_CASE		)

									# class embedding
									if class_embed_type is None and num_class_embeds is not None:
																_UpperCAmelCase					=    nn.Embedding(_SCREAMING_SNAKE_CASE    ,					_SCREAMING_SNAKE_CASE		)
									elif class_embed_type == "timestep":
																_UpperCAmelCase					=    TimestepEmbedding(_SCREAMING_SNAKE_CASE    ,					_SCREAMING_SNAKE_CASE		)
									elif class_embed_type == "identity":
																_UpperCAmelCase					=    nn.Identity(_SCREAMING_SNAKE_CASE    ,					_SCREAMING_SNAKE_CASE		)
									else:
																_UpperCAmelCase					=    None

									_UpperCAmelCase					=    nn.ModuleList([]		)
									_UpperCAmelCase					=    None
									_UpperCAmelCase					=    nn.ModuleList([]		)

									# down
									_UpperCAmelCase					=    block_out_channels[0]
									for i, down_block_type in enumerate(_SCREAMING_SNAKE_CASE		):
																_UpperCAmelCase					=    output_channel
																_UpperCAmelCase					=    block_out_channels[i]
																_UpperCAmelCase					=    i == len(_SCREAMING_SNAKE_CASE		) - 1

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

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

									# up
									_UpperCAmelCase					=    list(reversed(_SCREAMING_SNAKE_CASE		)		)
									_UpperCAmelCase					=    reversed_block_out_channels[0]
									for i, up_block_type in enumerate(_SCREAMING_SNAKE_CASE		):
																_UpperCAmelCase					=    output_channel
																_UpperCAmelCase					=    reversed_block_out_channels[i]
																_UpperCAmelCase					=    reversed_block_out_channels[min(i + 1    ,					len(_SCREAMING_SNAKE_CASE		) - 1		)]

																_UpperCAmelCase					=    i == len(_SCREAMING_SNAKE_CASE		) - 1

																_UpperCAmelCase					=    get_up_block(
																    _SCREAMING_SNAKE_CASE    ,					num_layers=layers_per_block + 1    ,					in_channels=_SCREAMING_SNAKE_CASE    ,					out_channels=_SCREAMING_SNAKE_CASE    ,					prev_output_channel=_SCREAMING_SNAKE_CASE    ,					temb_channels=_SCREAMING_SNAKE_CASE    ,					add_upsample=not is_final_block    ,					resnet_eps=_SCREAMING_SNAKE_CASE    ,					resnet_act_fn=_SCREAMING_SNAKE_CASE    ,					resnet_groups=_SCREAMING_SNAKE_CASE    ,					attention_head_dim=attention_head_dim if attention_head_dim is not None else output_channel    ,					resnet_time_scale_shift=_SCREAMING_SNAKE_CASE    ,					upsample_type=_SCREAMING_SNAKE_CASE    ,					)
																self.up_blocks.append(_SCREAMING_SNAKE_CASE		)
																_UpperCAmelCase					=    output_channel

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






		def 			UpperCAmelCase__       (    self    ,					_SCREAMING_SNAKE_CASE    ,					_SCREAMING_SNAKE_CASE    ,					_SCREAMING_SNAKE_CASE = None    ,					_SCREAMING_SNAKE_CASE = True    ,					)				-> Union[UNetaDOutput, Tuple]:




									"""simple docstring"""





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

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

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

									_UpperCAmelCase					=    self.time_proj(_SCREAMING_SNAKE_CASE		)

									# 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.
									_UpperCAmelCase					=    t_emb.to(dtype=self.dtype		)
									_UpperCAmelCase					=    self.time_embedding(_SCREAMING_SNAKE_CASE		)

									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":
																							_UpperCAmelCase					=    self.time_proj(_SCREAMING_SNAKE_CASE		)

																_UpperCAmelCase					=    self.class_embedding(_SCREAMING_SNAKE_CASE		).to(dtype=self.dtype		)
																_UpperCAmelCase					=    emb + class_emb

									# 2. pre-process
									_UpperCAmelCase					=    sample
									_UpperCAmelCase					=    self.conv_in(_SCREAMING_SNAKE_CASE		)

									# 3. down
									_UpperCAmelCase					=    (sample,)
									for downsample_block in self.down_blocks:
																if hasattr(_SCREAMING_SNAKE_CASE    ,					'skip_conv'		):
																							_UpperCAmelCase	, _UpperCAmelCase	, _UpperCAmelCase					=    downsample_block(
																							    hidden_states=_SCREAMING_SNAKE_CASE    ,					temb=_SCREAMING_SNAKE_CASE    ,					skip_sample=_SCREAMING_SNAKE_CASE		)
																else:
																							_UpperCAmelCase	, _UpperCAmelCase					=    downsample_block(hidden_states=_SCREAMING_SNAKE_CASE    ,					temb=_SCREAMING_SNAKE_CASE		)

																down_block_res_samples += res_samples

									# 4. mid
									_UpperCAmelCase					=    self.mid_block(_SCREAMING_SNAKE_CASE    ,					_SCREAMING_SNAKE_CASE		)

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

																if hasattr(_SCREAMING_SNAKE_CASE    ,					'skip_conv'		):
																							_UpperCAmelCase	, _UpperCAmelCase					=    upsample_block(_SCREAMING_SNAKE_CASE    ,					_SCREAMING_SNAKE_CASE    ,					_SCREAMING_SNAKE_CASE    ,					_SCREAMING_SNAKE_CASE		)
																else:
																							_UpperCAmelCase					=    upsample_block(_SCREAMING_SNAKE_CASE    ,					_SCREAMING_SNAKE_CASE    ,					_SCREAMING_SNAKE_CASE		)

        # 6. post-process
									_UpperCAmelCase					=    self.conv_norm_out(_SCREAMING_SNAKE_CASE		)
									_UpperCAmelCase					=    self.conv_act(_SCREAMING_SNAKE_CASE		)
									_UpperCAmelCase					=    self.conv_out(_SCREAMING_SNAKE_CASE		)

									if skip_sample is not None:
																sample += skip_sample

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

									if not return_dict:
																return (sample,)

									return UNetaDOutput(sample=_SCREAMING_SNAKE_CASE		)




import warnings

from ...processing_utils import ProcessorMixin
from ...tokenization_utils_base import BatchEncoding

class 	__a   (       UpperCAmelCase    ):
		_a			:						int    			=			['image_processor', 'tokenizer']
		_a			:						Union[str, Any]    			=			'AutoImageProcessor'
		_a			:						Optional[int]    			=			'AutoTokenizer'
		def __init__(    self    ,					_SCREAMING_SNAKE_CASE=None    ,					_SCREAMING_SNAKE_CASE=None    ,					**_SCREAMING_SNAKE_CASE		)				-> Tuple:




									"""simple docstring"""





									_UpperCAmelCase					=    None
									if "feature_extractor" in kwargs:
																warnings.warn(
																    'The `feature_extractor` argument is deprecated and will be removed in v5, use `image_processor`'
																    ' instead.'    ,					_SCREAMING_SNAKE_CASE    ,					)
																_UpperCAmelCase					=    kwargs.pop('feature_extractor'		)

									_UpperCAmelCase					=    image_processor if image_processor is not None else feature_extractor
									if image_processor is None:
																raise ValueError('You have to specify an image_processor.'		)
									if tokenizer is None:
																raise ValueError('You have to specify a tokenizer.'		)

									super().__init__(_SCREAMING_SNAKE_CASE    ,					_SCREAMING_SNAKE_CASE		)
									_UpperCAmelCase					=    self.image_processor
		def __call__(    self    ,					_SCREAMING_SNAKE_CASE=None    ,					_SCREAMING_SNAKE_CASE=None    ,					_SCREAMING_SNAKE_CASE=None    ,					**_SCREAMING_SNAKE_CASE		)				-> Dict:




									"""simple docstring"""





									if text is None and images is None:
																raise ValueError('You have to specify either text or images. Both cannot be none.'		)

									if text is not None:
																_UpperCAmelCase					=    self.tokenizer(_SCREAMING_SNAKE_CASE    ,					return_tensors=_SCREAMING_SNAKE_CASE    ,					**_SCREAMING_SNAKE_CASE		)

									if images is not None:
																_UpperCAmelCase					=    self.image_processor(_SCREAMING_SNAKE_CASE    ,					return_tensors=_SCREAMING_SNAKE_CASE    ,					**_SCREAMING_SNAKE_CASE		)

									if text is not None and images is not None:
																_UpperCAmelCase					=    image_features.pixel_values
																return encoding
									elif text is not None:
																return encoding
									else:
																return BatchEncoding(data=dict(**_SCREAMING_SNAKE_CASE		)    ,					tensor_type=_SCREAMING_SNAKE_CASE		)
		def 			UpperCAmelCase__       (    self    ,					*_SCREAMING_SNAKE_CASE    ,					**_SCREAMING_SNAKE_CASE		)				-> int:




									"""simple docstring"""





									return self.tokenizer.batch_decode(*_SCREAMING_SNAKE_CASE    ,					**_SCREAMING_SNAKE_CASE		)
		def 			UpperCAmelCase__       (    self    ,					*_SCREAMING_SNAKE_CASE    ,					**_SCREAMING_SNAKE_CASE		)				-> Tuple:




									"""simple docstring"""





									return self.tokenizer.decode(*_SCREAMING_SNAKE_CASE    ,					**_SCREAMING_SNAKE_CASE		)
		@property
		def 			UpperCAmelCase__       (    self		)				-> Optional[int]:




									"""simple docstring"""





									_UpperCAmelCase					=    self.tokenizer.model_input_names
									_UpperCAmelCase					=    self.image_processor.model_input_names
									return list(dict.fromkeys(tokenizer_input_names + image_processor_input_names		)		)
		@property
		def 			UpperCAmelCase__       (    self		)				-> Any:




									"""simple docstring"""





									warnings.warn(
									    '`feature_extractor_class` is deprecated and will be removed in v5. Use `image_processor_class` instead.'    ,					_SCREAMING_SNAKE_CASE    ,					)
									return self.image_processor_class






		@property
		def 			UpperCAmelCase__       (    self		)				-> Any:




									"""simple docstring"""





									warnings.warn(
									    '`feature_extractor` is deprecated and will be removed in v5. Use `image_processor` instead.'    ,					_SCREAMING_SNAKE_CASE    ,					)
									return self.image_processor
