



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

import torch
import torch.nn as nn

from ..configuration_utils import ConfigMixin, register_to_config
from ..utils import BaseOutput, apply_forward_hook
from .attention_processor import AttentionProcessor, AttnProcessor
from .modeling_utils import ModelMixin
from .vae import Decoder, DecoderOutput, DiagonalGaussianDistribution, Encoder

@dataclass
class   _A   ( _lowercase	):


						'''simple docstring'''





						_snake_case			:     "DiagonalGaussianDistribution"

class   _A   ( _lowercase  ,						_lowercase	):


						'''simple docstring'''





						_snake_case			:     Optional[Any]      							=      True



						@register_to_config
						def __init__(	self	:							List[str]							,      lowerCamelCase	:							int = 3							,      lowerCamelCase	:							int = 3							,      lowerCamelCase	:							Tuple[str] = ("DownEncoderBlock2D",)							,      lowerCamelCase	:							Tuple[str] = ("UpDecoderBlock2D",)							,      lowerCamelCase	:							Tuple[int] = (64,)							,      lowerCamelCase	:							int = 1							,      lowerCamelCase	:							str = "silu"							,      lowerCamelCase	:							int = 4							,      lowerCamelCase	:							int = 32							,      lowerCamelCase	:							int = 32							,      lowerCamelCase	:							float = 0.1_8215							,      ):






							'''simple docstring'''



							super().__init__()

							# pass init params to Encoder
							__lowercase		       =			Encoder(
							    in_channels=lowerCamelCase							,      out_channels=lowerCamelCase							,      down_block_types=lowerCamelCase							,      block_out_channels=lowerCamelCase							,      layers_per_block=lowerCamelCase							,      act_fn=lowerCamelCase							,      norm_num_groups=lowerCamelCase							,      double_z=lowerCamelCase							,      )

							# pass init params to Decoder
							__lowercase		       =			Decoder(
							    in_channels=lowerCamelCase							,      out_channels=lowerCamelCase							,      up_block_types=lowerCamelCase							,      block_out_channels=lowerCamelCase							,      layers_per_block=lowerCamelCase							,      norm_num_groups=lowerCamelCase							,      act_fn=lowerCamelCase							,      )

							__lowercase		       =			nn.Convad(2 * latent_channels							,      2 * latent_channels							,      1	)
							__lowercase		       =			nn.Convad(lowerCamelCase							,      lowerCamelCase							,      1	)

							__lowercase		       =			False
							__lowercase		       =			False

							# only relevant if vae tiling is enabled
							__lowercase		       =			self.config.sample_size
							__lowercase		       =			(
							    self.config.sample_size[0]
							    if isinstance(self.config.sample_size							,      (list, tuple)	)
							    else self.config.sample_size
							)
							__lowercase		       =			int(sample_size / (2 ** (len(self.config.block_out_channels	) - 1))	)
							__lowercase		       =			0.25



						def     _snake_case	(	self	:							Any							,      lowerCamelCase	:							int							,      lowerCamelCase	:							Dict=False	):






							'''simple docstring'''



							if isinstance(lowerCamelCase							,      (Encoder, Decoder)	):
								__lowercase		       =			value



						def     _snake_case	(	self	:							Tuple							,      lowerCamelCase	:							bool = True	):






							'''simple docstring'''



							__lowercase		       =			use_tiling



						def     _snake_case	(	self	:							Any	):






							'''simple docstring'''



							self.enable_tiling(lowerCamelCase	)



						def     _snake_case	(	self	:							Any	):






							'''simple docstring'''



							__lowercase		       =			True



						def     _snake_case	(	self	:							Optional[Any]	):






							'''simple docstring'''



							__lowercase		       =			False



						@property
						# Copied from diffusers.models.unet_2d_condition.UNet2DConditionModel.attn_processors
						def     _snake_case	(	self	:							Tuple	):






							'''simple docstring'''



							__lowercase		       =			{}

							def fn_recursive_add_processors(lowerCamelCase	:							str							,      lowerCamelCase	:							torch.nn.Module							,      lowerCamelCase	:							Dict[str, AttentionProcessor]	):
								if hasattr(lowerCamelCase							,      "set_processor"	):
									__lowercase		       =			module.processor

								for sub_name, child in module.named_children():
									fn_recursive_add_processors(f"""{name}.{sub_name}"""							,      lowerCamelCase							,      lowerCamelCase	)

								return processors

							for name, module in self.named_children():
								fn_recursive_add_processors(lowerCamelCase							,      lowerCamelCase							,      lowerCamelCase	)

							return processors



						def     _snake_case	(	self	:							str							,      lowerCamelCase	:							Union[AttentionProcessor, Dict[str, AttentionProcessor]]	):






							'''simple docstring'''



							__lowercase		       =			len(self.attn_processors.keys()	)

							if isinstance(lowerCamelCase							,      lowerCamelCase	) and len(lowerCamelCase	) != count:
								raise ValueError(
								    f"""A dict of processors was passed, but the number of processors {len(lowerCamelCase	)} does not match the"""
								    f""" number of attention layers: {count}. Please make sure to pass {count} processor classes."""	)

							def fn_recursive_attn_processor(lowerCamelCase	:							str							,      lowerCamelCase	:							torch.nn.Module							,      lowerCamelCase	:							Any	):
								if hasattr(lowerCamelCase							,      "set_processor"	):
									if not isinstance(lowerCamelCase							,      lowerCamelCase	):
										module.set_processor(lowerCamelCase	)
									else:
										module.set_processor(processor.pop(f"""{name}.processor"""	)	)

								for sub_name, child in module.named_children():
									fn_recursive_attn_processor(f"""{name}.{sub_name}"""							,      lowerCamelCase							,      lowerCamelCase	)

							for name, module in self.named_children():
								fn_recursive_attn_processor(lowerCamelCase							,      lowerCamelCase							,      lowerCamelCase	)



						def     _snake_case	(	self	:							int	):






							'''simple docstring'''



							self.set_attn_processor(AttnProcessor()	)



						@apply_forward_hook
						def     _snake_case	(	self	:							Tuple							,      lowerCamelCase	:							torch.FloatTensor							,      lowerCamelCase	:							bool = True	):






							'''simple docstring'''



							if self.use_tiling and (x.shape[-1] > self.tile_sample_min_size or x.shape[-2] > self.tile_sample_min_size):
								return self.tiled_encode(lowerCamelCase							,      return_dict=lowerCamelCase	)

							if self.use_slicing and x.shape[0] > 1:
								__lowercase		       =			[self.encoder(lowerCamelCase	) for x_slice in x.split(1	)]
								__lowercase		       =			torch.cat(lowerCamelCase	)
							else:
								__lowercase		       =			self.encoder(lowerCamelCase	)

							__lowercase		       =			self.quant_conv(lowerCamelCase	)
							__lowercase		       =			DiagonalGaussianDistribution(lowerCamelCase	)

							if not return_dict:
								return (posterior,)

							return AutoencoderKLOutput(latent_dist=lowerCamelCase	)



						def     _snake_case	(	self	:							str							,      lowerCamelCase	:							torch.FloatTensor							,      lowerCamelCase	:							bool = True	):






							'''simple docstring'''



							if self.use_tiling and (z.shape[-1] > self.tile_latent_min_size or z.shape[-2] > self.tile_latent_min_size):
								return self.tiled_decode(lowerCamelCase							,      return_dict=lowerCamelCase	)

							__lowercase		       =			self.post_quant_conv(lowerCamelCase	)
							__lowercase		       =			self.decoder(lowerCamelCase	)

							if not return_dict:
								return (dec,)

							return DecoderOutput(sample=lowerCamelCase	)



						@apply_forward_hook
						def     _snake_case	(	self	:							Optional[Any]							,      lowerCamelCase	:							torch.FloatTensor							,      lowerCamelCase	:							bool = True	):






							'''simple docstring'''



							if self.use_slicing and z.shape[0] > 1:
								__lowercase		       =			[self._decode(lowerCamelCase	).sample for z_slice in z.split(1	)]
								__lowercase		       =			torch.cat(lowerCamelCase	)
							else:
								__lowercase		       =			self._decode(lowerCamelCase	).sample

							if not return_dict:
								return (decoded,)

							return DecoderOutput(sample=lowerCamelCase	)



						def     _snake_case	(	self	:							List[Any]							,      lowerCamelCase	:							Optional[Any]							,      lowerCamelCase	:							Optional[int]							,      lowerCamelCase	:							List[Any]	):






							'''simple docstring'''



							__lowercase		       =			min(a.shape[2]							,      b.shape[2]							,      lowerCamelCase	)
							for y in range(lowerCamelCase	):
								__lowercase		       =			a[:, :, -blend_extent + y, :] * (1 - y / blend_extent) + b[:, :, y, :] * (y / blend_extent)
							return b



						def     _snake_case	(	self	:							Optional[int]							,      lowerCamelCase	:							Union[str, Any]							,      lowerCamelCase	:							Union[str, Any]							,      lowerCamelCase	:							List[Any]	):






							'''simple docstring'''



							__lowercase		       =			min(a.shape[3]							,      b.shape[3]							,      lowerCamelCase	)
							for x in range(lowerCamelCase	):
								__lowercase		       =			a[:, :, :, -blend_extent + x] * (1 - x / blend_extent) + b[:, :, :, x] * (x / blend_extent)
							return b



						def     _snake_case	(	self	:							List[Any]							,      lowerCamelCase	:							torch.FloatTensor							,      lowerCamelCase	:							bool = True	):






							'''simple docstring'''



							__lowercase		       =			int(self.tile_sample_min_size * (1 - self.tile_overlap_factor)	)
							__lowercase		       =			int(self.tile_latent_min_size * self.tile_overlap_factor	)
							__lowercase		       =			self.tile_latent_min_size - blend_extent

							# Split the image into 512x512 tiles and encode them separately.
							__lowercase		       =			[]
							for i in range(0							,      x.shape[2]							,      lowerCamelCase	):
								__lowercase		       =			[]
								for j in range(0							,      x.shape[3]							,      lowerCamelCase	):
									__lowercase		       =			x[:, :, i : i + self.tile_sample_min_size, j : j + self.tile_sample_min_size]
									__lowercase		       =			self.encoder(lowerCamelCase	)
									__lowercase		       =			self.quant_conv(lowerCamelCase	)
									row.append(lowerCamelCase	)
								rows.append(lowerCamelCase	)
							__lowercase		       =			[]
							for i, row in enumerate(lowerCamelCase	):
								__lowercase		       =			[]
								for j, tile in enumerate(lowerCamelCase	):
									# blend the above tile and the left tile
									# to the current tile and add the current tile to the result row
									if i > 0:
										__lowercase		       =			self.blend_v(rows[i - 1][j]							,      lowerCamelCase							,      lowerCamelCase	)
									if j > 0:
										__lowercase		       =			self.blend_h(row[j - 1]							,      lowerCamelCase							,      lowerCamelCase	)
									result_row.append(tile[:, :, :row_limit, :row_limit]	)
								result_rows.append(torch.cat(lowerCamelCase							,      dim=3	)	)

							__lowercase		       =			torch.cat(lowerCamelCase							,      dim=2	)
							__lowercase		       =			DiagonalGaussianDistribution(lowerCamelCase	)

							if not return_dict:
								return (posterior,)

							return AutoencoderKLOutput(latent_dist=lowerCamelCase	)



						def     _snake_case	(	self	:							Dict							,      lowerCamelCase	:							torch.FloatTensor							,      lowerCamelCase	:							bool = True	):






							'''simple docstring'''



							__lowercase		       =			int(self.tile_latent_min_size * (1 - self.tile_overlap_factor)	)
							__lowercase		       =			int(self.tile_sample_min_size * self.tile_overlap_factor	)
							__lowercase		       =			self.tile_sample_min_size - blend_extent

							# Split z into overlapping 64x64 tiles and decode them separately.
							# The tiles have an overlap to avoid seams between tiles.
							__lowercase		       =			[]
							for i in range(0							,      z.shape[2]							,      lowerCamelCase	):
								__lowercase		       =			[]
								for j in range(0							,      z.shape[3]							,      lowerCamelCase	):
									__lowercase		       =			z[:, :, i : i + self.tile_latent_min_size, j : j + self.tile_latent_min_size]
									__lowercase		       =			self.post_quant_conv(lowerCamelCase	)
									__lowercase		       =			self.decoder(lowerCamelCase	)
									row.append(lowerCamelCase	)
								rows.append(lowerCamelCase	)
							__lowercase		       =			[]
							for i, row in enumerate(lowerCamelCase	):
								__lowercase		       =			[]
								for j, tile in enumerate(lowerCamelCase	):
									# blend the above tile and the left tile
									# to the current tile and add the current tile to the result row
									if i > 0:
										__lowercase		       =			self.blend_v(rows[i - 1][j]							,      lowerCamelCase							,      lowerCamelCase	)
									if j > 0:
										__lowercase		       =			self.blend_h(row[j - 1]							,      lowerCamelCase							,      lowerCamelCase	)
									result_row.append(tile[:, :, :row_limit, :row_limit]	)
								result_rows.append(torch.cat(lowerCamelCase							,      dim=3	)	)

							__lowercase		       =			torch.cat(lowerCamelCase							,      dim=2	)
							if not return_dict:
								return (dec,)

							return DecoderOutput(sample=lowerCamelCase	)



						def     _snake_case	(	self	:							Dict							,      lowerCamelCase	:							torch.FloatTensor							,      lowerCamelCase	:							bool = False							,      lowerCamelCase	:							bool = True							,      lowerCamelCase	:							Optional[torch.Generator] = None							,      ):






							'''simple docstring'''



							__lowercase		       =			sample
							__lowercase		       =			self.encode(lowerCamelCase	).latent_dist
							if sample_posterior:
								__lowercase		       =			posterior.sample(generator=lowerCamelCase	)
							else:
								__lowercase		       =			posterior.mode()
							__lowercase		       =			self.decode(lowerCamelCase	).sample

							if not return_dict:
								return (dec,)

							return DecoderOutput(sample=lowerCamelCase	)








import torch
from torch import nn

from ...configuration_utils import ConfigMixin, register_to_config
from ...models import ModelMixin

class   _A   ( _lowercase  ,						_lowercase	):


						'''simple docstring'''





						@register_to_config
						def __init__(	self	:							Optional[Any]							,      *,
						    lowerCamelCase	:							int = 4							,      lowerCamelCase	:							int = 768							,      lowerCamelCase	:							int							,      lowerCamelCase	:							Optional[int]							,      ):






							'''simple docstring'''



							super().__init__()

							__lowercase		       =			nn.Parameter(torch.zeros(lowerCamelCase	)	)

							# parameters for additional clip time embeddings
							__lowercase		       =			nn.Linear(lowerCamelCase							,      lowerCamelCase	)
							__lowercase		       =			nn.Linear(lowerCamelCase							,      lowerCamelCase	)

							# parameters for encoder hidden states
							__lowercase		       =			clip_extra_context_tokens
							__lowercase		       =			nn.Linear(
							    lowerCamelCase							,      self.clip_extra_context_tokens * cross_attention_dim	)
							__lowercase		       =			nn.Linear(lowerCamelCase							,      lowerCamelCase	)
							__lowercase		       =			nn.LayerNorm(lowerCamelCase	)



						def     _snake_case	(	self	:							Union[str, Any]							,      *, lowerCamelCase	:							Any							,      lowerCamelCase	:							Tuple							,      lowerCamelCase	:							Optional[int]							,      lowerCamelCase	:							Tuple	):






							'''simple docstring'''



							if do_classifier_free_guidance:
								# Add the classifier free guidance embeddings to the image embeddings
								__lowercase		       =			image_embeddings.shape[0]
								__lowercase		       =			self.learned_classifier_free_guidance_embeddings.unsqueeze(0	)
								__lowercase		       =			classifier_free_guidance_embeddings.expand(
								    lowerCamelCase							,      -1	)
								__lowercase		       =			torch.cat([classifier_free_guidance_embeddings, image_embeddings]							,      dim=0	)

							# The image embeddings batch size and the text embeddings batch size are equal
							assert image_embeddings.shape[0] == prompt_embeds.shape[0]

							__lowercase		       =			prompt_embeds.shape[0]

							# "Specifically, we modify the architecture described in Nichol et al. (2021) by projecting and
							# adding CLIP embeddings to the existing timestep embedding, ...
							__lowercase		       =			self.embedding_proj(lowerCamelCase	)
							__lowercase		       =			self.clip_image_embeddings_project_to_time_embeddings(lowerCamelCase	)
							__lowercase		       =			time_projected_image_embeddings + time_projected_prompt_embeds

							# ... and by projecting CLIP embeddings into four
							# extra tokens of context that are concatenated to the sequence of outputs from the GLIDE text encoder"
							__lowercase		       =			self.clip_extra_context_tokens_proj(lowerCamelCase	)
							__lowercase		       =			clip_extra_context_tokens.reshape(lowerCamelCase							,      -1							,      self.clip_extra_context_tokens	)
							__lowercase		       =			clip_extra_context_tokens.permute(0							,      2							,      1	)

							__lowercase		       =			self.encoder_hidden_states_proj(lowerCamelCase	)
							__lowercase		       =			self.text_encoder_hidden_states_norm(lowerCamelCase	)
							__lowercase		       =			torch.cat([clip_extra_context_tokens, text_encoder_hidden_states]							,      dim=1	)

							return text_encoder_hidden_states, additive_clip_time_embeddings



