

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 _UpperCamelCase				(				__A ):
			'''simple docstring'''







			lowerCamelCase__							     =42






class _UpperCamelCase				(				__A				,						__A ):
			'''simple docstring'''







			lowerCamelCase__							     =True
			@register_to_config
			def __init__(      self				:			Dict   ,   a				:			int = 3   ,   a				:			int = 3   ,   a				:			Tuple[str] = ("DownEncoderBlock2D",)   ,   a				:			Tuple[str] = ("UpDecoderBlock2D",)   ,   a				:			Tuple[int] = (64,)   ,   a				:			int = 1   ,   a				:			str = "silu"   ,   a				:			int = 4   ,   a				:			int = 32   ,   a				:			int = 32   ,   a				:			float = 0.1_8215   ,   )				->							Union[str, Any]:

				"""simple docstring"""

				super().__init__()

				# pass init params to Encoder
				SCREAMING_SNAKE_CASE :   List[Any]					  =							Encoder(
				    in_channels=a   ,   out_channels=a   ,   down_block_types=a   ,   block_out_channels=a   ,   layers_per_block=a   ,   act_fn=a   ,   norm_num_groups=a   ,   double_z=a   ,   )

				# pass init params to Decoder
				SCREAMING_SNAKE_CASE :   Dict					  =							Decoder(
				    in_channels=a   ,   out_channels=a   ,   up_block_types=a   ,   block_out_channels=a   ,   layers_per_block=a   ,   norm_num_groups=a   ,   act_fn=a   ,   )

				SCREAMING_SNAKE_CASE :   Any					  =							nn.Convad(2 * latent_channels   ,   2 * latent_channels   ,   1						)
				SCREAMING_SNAKE_CASE :   Dict					  =							nn.Convad(a   ,   a   ,   1						)

				SCREAMING_SNAKE_CASE :   Any					  =							False
				SCREAMING_SNAKE_CASE :   Optional[Any]					  =							False

				# only relevant if vae tiling is enabled
				SCREAMING_SNAKE_CASE :   Dict					  =							self.config.sample_size
				SCREAMING_SNAKE_CASE :   int					  =							(
				    self.config.sample_size[0]
				    if isinstance(self.config.sample_size   ,   (list, tuple)						)
				    else self.config.sample_size
				)
				SCREAMING_SNAKE_CASE :   Union[str, Any]					  =							int(sample_size / (2 ** (len(self.config.block_out_channels						) - 1))						)
				SCREAMING_SNAKE_CASE :   Union[str, Any]					  =							0.25
			def 				__UpperCamelCase			(      self				:			Optional[Any]   ,   a				:			Dict   ,   a				:			Tuple=False						)				->							List[str]:

				"""simple docstring"""

				if isinstance(a   ,   (Encoder, Decoder)						):
					SCREAMING_SNAKE_CASE :   Any					  =							value
			def 				__UpperCamelCase			(      self				:			Tuple   ,   a				:			bool = True						)				->							str:

				"""simple docstring"""

				SCREAMING_SNAKE_CASE :   Optional[int]					  =							use_tiling
			def 				__UpperCamelCase			(      self				:			Any						)				->							str:

				"""simple docstring"""

				self.enable_tiling(a						)
			def 				__UpperCamelCase			(      self				:			List[str]						)				->							Dict:

				"""simple docstring"""

				SCREAMING_SNAKE_CASE :   Tuple					  =							True
			def 				__UpperCamelCase			(      self				:			List[str]						)				->							int:

				"""simple docstring"""

				SCREAMING_SNAKE_CASE :   str					  =							False
			@property
			# Copied from diffusers.models.unet_2d_condition.UNet2DConditionModel.attn_processors
			def 				__UpperCamelCase			(      self				:			Any						)				->							Dict[str, AttentionProcessor]:

				"""simple docstring"""

				SCREAMING_SNAKE_CASE :   Any					  =							{}

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

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

					return processors

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

				return processors
			def 				__UpperCamelCase			(      self				:			Tuple   ,   a				:			Union[AttentionProcessor, Dict[str, AttentionProcessor]]						)				->							Optional[Any]:

				"""simple docstring"""

				SCREAMING_SNAKE_CASE :   Optional[Any]					  =							len(self.attn_processors.keys()						)

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

				def fn_recursive_attn_processor(a				:			str   ,   a				:			torch.nn.Module   ,   a				:			Any						):
					if hasattr(a   ,   "set_processor"						):
						if not isinstance(a   ,   a						):
							module.set_processor(a						)
						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}"   ,   a   ,   a						)

				for name, module in self.named_children():
					fn_recursive_attn_processor(a   ,   a   ,   a						)
			def 				__UpperCamelCase			(      self				:			str						)				->							Dict:

				"""simple docstring"""

				self.set_attn_processor(AttnProcessor()						)
			@apply_forward_hook
			def 				__UpperCamelCase			(      self				:			Optional[Any]   ,   a				:			torch.FloatTensor   ,   a				:			bool = True						)				->							AutoencoderKLOutput:

				"""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(a   ,   return_dict=a						)

				if self.use_slicing and x.shape[0] > 1:
					SCREAMING_SNAKE_CASE :   str					  =							[self.encoder(a						) for x_slice in x.split(1						)]
					SCREAMING_SNAKE_CASE :   Any					  =							torch.cat(a						)
				else:
					SCREAMING_SNAKE_CASE :   Optional[Any]					  =							self.encoder(a						)

				SCREAMING_SNAKE_CASE :   Any					  =							self.quant_conv(a						)
				SCREAMING_SNAKE_CASE :   Optional[Any]					  =							DiagonalGaussianDistribution(a						)

				if not return_dict:
					return (posterior,)

				return AutoencoderKLOutput(latent_dist=a						)
			def 				__UpperCamelCase			(      self				:			List[str]   ,   a				:			torch.FloatTensor   ,   a				:			bool = True						)				->							Union[DecoderOutput, torch.FloatTensor]:

				"""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(a   ,   return_dict=a						)

				SCREAMING_SNAKE_CASE :   int					  =							self.post_quant_conv(a						)
				SCREAMING_SNAKE_CASE :   List[Any]					  =							self.decoder(a						)

				if not return_dict:
					return (dec,)

				return DecoderOutput(sample=a						)
			@apply_forward_hook
			def 				__UpperCamelCase			(      self				:			str   ,   a				:			torch.FloatTensor   ,   a				:			bool = True						)				->							Union[DecoderOutput, torch.FloatTensor]:

				"""simple docstring"""

				if self.use_slicing and z.shape[0] > 1:
					SCREAMING_SNAKE_CASE :   List[str]					  =							[self._decode(a						).sample for z_slice in z.split(1						)]
					SCREAMING_SNAKE_CASE :   Any					  =							torch.cat(a						)
				else:
					SCREAMING_SNAKE_CASE :   Optional[int]					  =							self._decode(a						).sample

				if not return_dict:
					return (decoded,)

				return DecoderOutput(sample=a						)
			def 				__UpperCamelCase			(      self				:			List[Any]   ,   a				:			Union[str, Any]   ,   a				:			Optional[int]   ,   a				:			int						)				->							str:

				"""simple docstring"""

				SCREAMING_SNAKE_CASE :   List[str]					  =							min(a.shape[2]   ,   b.shape[2]   ,   a						)
				for y in range(a						):
					SCREAMING_SNAKE_CASE :   List[str]					  =							a[:, :, -blend_extent + y, :] * (1 - y / blend_extent) + b[:, :, y, :] * (y / blend_extent)
				return b
			def 				__UpperCamelCase			(      self				:			Any   ,   a				:			Optional[int]   ,   a				:			Optional[Any]   ,   a				:			Tuple						)				->							str:

				"""simple docstring"""

				SCREAMING_SNAKE_CASE :   List[str]					  =							min(a.shape[3]   ,   b.shape[3]   ,   a						)
				for x in range(a						):
					SCREAMING_SNAKE_CASE :   Tuple					  =							a[:, :, :, -blend_extent + x] * (1 - x / blend_extent) + b[:, :, :, x] * (x / blend_extent)
				return b
			def 				__UpperCamelCase			(      self				:			Dict   ,   a				:			torch.FloatTensor   ,   a				:			bool = True						)				->							AutoencoderKLOutput:

				"""simple docstring"""

				SCREAMING_SNAKE_CASE :   Dict					  =							int(self.tile_sample_min_size * (1 - self.tile_overlap_factor)						)
				SCREAMING_SNAKE_CASE :   Optional[Any]					  =							int(self.tile_latent_min_size * self.tile_overlap_factor						)
				SCREAMING_SNAKE_CASE :   Dict					  =							self.tile_latent_min_size - blend_extent

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

				SCREAMING_SNAKE_CASE :   List[str]					  =							torch.cat(a   ,   dim=2						)
				SCREAMING_SNAKE_CASE :   List[str]					  =							DiagonalGaussianDistribution(a						)

				if not return_dict:
					return (posterior,)

				return AutoencoderKLOutput(latent_dist=a						)
			def 				__UpperCamelCase			(      self				:			Optional[int]   ,   a				:			torch.FloatTensor   ,   a				:			bool = True						)				->							Union[DecoderOutput, torch.FloatTensor]:

				"""simple docstring"""

				SCREAMING_SNAKE_CASE :   Dict					  =							int(self.tile_latent_min_size * (1 - self.tile_overlap_factor)						)
				SCREAMING_SNAKE_CASE :   List[str]					  =							int(self.tile_sample_min_size * self.tile_overlap_factor						)
				SCREAMING_SNAKE_CASE :   Dict					  =							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.
				SCREAMING_SNAKE_CASE :   str					  =							[]
				for i in range(0   ,   z.shape[2]   ,   a						):
					SCREAMING_SNAKE_CASE :   Optional[Any]					  =							[]
					for j in range(0   ,   z.shape[3]   ,   a						):
						SCREAMING_SNAKE_CASE :   str					  =							z[:, :, i : i + self.tile_latent_min_size, j : j + self.tile_latent_min_size]
						SCREAMING_SNAKE_CASE :   Optional[int]					  =							self.post_quant_conv(a						)
						SCREAMING_SNAKE_CASE :   List[Any]					  =							self.decoder(a						)
						row.append(a						)
					rows.append(a						)
				SCREAMING_SNAKE_CASE :   Any					  =							[]
				for i, row in enumerate(a						):
					SCREAMING_SNAKE_CASE :   Optional[int]					  =							[]
					for j, tile in enumerate(a						):
						# blend the above tile and the left tile
						# to the current tile and add the current tile to the result row
						if i > 0:
							SCREAMING_SNAKE_CASE :   List[str]					  =							self.blend_v(rows[i - 1][j]   ,   a   ,   a						)
						if j > 0:
							SCREAMING_SNAKE_CASE :   List[str]					  =							self.blend_h(row[j - 1]   ,   a   ,   a						)
						result_row.append(tile[:, :, :row_limit, :row_limit]						)
					result_rows.append(torch.cat(a   ,   dim=3						)						)

				SCREAMING_SNAKE_CASE :   Dict					  =							torch.cat(a   ,   dim=2						)
				if not return_dict:
					return (dec,)

				return DecoderOutput(sample=a						)






			def 				__UpperCamelCase			(      self				:			List[Any]   ,   a				:			torch.FloatTensor   ,   a				:			bool = False   ,   a				:			bool = True   ,   a				:			Optional[torch.Generator] = None   ,   )				->							Union[DecoderOutput, torch.FloatTensor]:

				"""simple docstring"""

				SCREAMING_SNAKE_CASE :   Dict					  =							sample
				SCREAMING_SNAKE_CASE :   Optional[int]					  =							self.encode(a						).latent_dist
				if sample_posterior:
					SCREAMING_SNAKE_CASE :   Optional[int]					  =							posterior.sample(generator=a						)
				else:
					SCREAMING_SNAKE_CASE :   Tuple					  =							posterior.mode()
				SCREAMING_SNAKE_CASE :   List[str]					  =							self.decode(a						).sample

				if not return_dict:
					return (dec,)

				return DecoderOutput(sample=a						)


a_      =     frozenset(
    [
        'prompt',
        'height',
        'width',
        'guidance_scale',
        'negative_prompt',
        'prompt_embeds',
        'negative_prompt_embeds',
        'cross_attention_kwargs',
    ]
)

a_      =     frozenset(['prompt', 'negative_prompt'])

a_      =     frozenset([])

a_      =     frozenset(['image'])

a_      =     frozenset(
    [
        'image',
        'height',
        'width',
        'guidance_scale',
    ]
)

a_      =     frozenset(['image'])

a_      =     frozenset(
    [
        'prompt',
        'image',
        'height',
        'width',
        'guidance_scale',
        'negative_prompt',
        'prompt_embeds',
        'negative_prompt_embeds',
    ]
)

a_      =     frozenset(['prompt', 'image', 'negative_prompt'])

a_      =     frozenset(
    [
        # Text guided image variation with an image mask
        'prompt',
        'image',
        'mask_image',
        'height',
        'width',
        'guidance_scale',
        'negative_prompt',
        'prompt_embeds',
        'negative_prompt_embeds',
    ]
)

a_      =     frozenset(['prompt', 'image', 'mask_image', 'negative_prompt'])

a_      =     frozenset(
    [
        # image variation with an image mask
        'image',
        'mask_image',
        'height',
        'width',
        'guidance_scale',
    ]
)

a_      =     frozenset(['image', 'mask_image'])

a_      =     frozenset(
    [
        'example_image',
        'image',
        'mask_image',
        'height',
        'width',
        'guidance_scale',
    ]
)

a_      =     frozenset(['example_image', 'image', 'mask_image'])

a_      =     frozenset(['class_labels'])

a_      =     frozenset(['class_labels'])

a_      =     frozenset(['batch_size'])

a_      =     frozenset([])

a_      =     frozenset(['batch_size'])

a_      =     frozenset([])

a_      =     frozenset(
    [
        'prompt',
        'audio_length_in_s',
        'guidance_scale',
        'negative_prompt',
        'prompt_embeds',
        'negative_prompt_embeds',
        'cross_attention_kwargs',
    ]
)

a_      =     frozenset(['prompt', 'negative_prompt'])
a_      =     frozenset(['input_tokens'])

a_      =     frozenset(['input_tokens'])