
"""simple docstring"""





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 __lowerCAmelCase					(       _lowercase   ):




			"""simple docstring"""







			snake_case						   =					42





class __lowerCAmelCase					(       _lowercase  ,					_lowercase   ):




			"""simple docstring"""







			snake_case						   =					True




			@register_to_config
			def __init__(					self    :       Any							, _snake_case    :       int = 3							, _snake_case    :       int = 3							, _snake_case    :       Tuple[str] = ("DownEncoderBlock2D",)							, _snake_case    :       Tuple[str] = ("UpDecoderBlock2D",)							, _snake_case    :       Tuple[int] = (64,)							, _snake_case    :       int = 1							, _snake_case    :       str = "silu"							, _snake_case    :       int = 4							, _snake_case    :       int = 32							, _snake_case    :       int = 32							, _snake_case    :       float = 0.1_8_2_1_5							, )   ->			Optional[Any]:







								"""simple docstring"""
								super().__init__()

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

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

								A_								=    nn.Convad(2 * latent_channels							, 2 * latent_channels							, 1						)
								A_								=    nn.Convad(_snake_case							, _snake_case							, 1						)

								A_								=    False
								A_								=    False

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




			def   lowerCamelCase__  (					self    :       str							, _snake_case    :       List[Any]							, _snake_case    :       Optional[int]=False						)   ->			str:







								"""simple docstring"""
								if isinstance(_snake_case							, (Encoder, Decoder)						):
													A_								=    value




			def   lowerCamelCase__  (					self    :       int							, _snake_case    :       bool = True						)   ->			Optional[int]:







								"""simple docstring"""
								A_								=    use_tiling




			def   lowerCamelCase__  (					self    :       List[str]						)   ->			Tuple:







								"""simple docstring"""
								self.enable_tiling(_snake_case						)




			def   lowerCamelCase__  (					self    :       List[Any]						)   ->			Optional[Any]:







								"""simple docstring"""
								A_								=    True




			def   lowerCamelCase__  (					self    :       str						)   ->			Tuple:







								"""simple docstring"""
								A_								=    False




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







								"""simple docstring"""
								A_								=    {}

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

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

													return processors

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

								return processors




			def   lowerCamelCase__  (					self    :       Tuple							, _snake_case    :       Union[AttentionProcessor, Dict[str, AttentionProcessor]]						)   ->			List[Any]:







								"""simple docstring"""
								A_								=    len(self.attn_processors.keys()						)

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

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

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




			def   lowerCamelCase__  (					self    :       str						)   ->			str:







								"""simple docstring"""
								self.set_attn_processor(AttnProcessor()						)




			@apply_forward_hook
			def   lowerCamelCase__  (					self    :       Optional[int]							, _snake_case    :       torch.FloatTensor							, _snake_case    :       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(_snake_case							, return_dict=_snake_case						)

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

								A_								=    self.quant_conv(_snake_case						)
								A_								=    DiagonalGaussianDistribution(_snake_case						)

								if not return_dict:
													return (posterior,)

								return AutoencoderKLOutput(latent_dist=_snake_case						)




			def   lowerCamelCase__  (					self    :       str							, _snake_case    :       torch.FloatTensor							, _snake_case    :       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(_snake_case							, return_dict=_snake_case						)

								A_								=    self.post_quant_conv(_snake_case						)
								A_								=    self.decoder(_snake_case						)

								if not return_dict:
													return (dec,)

								return DecoderOutput(sample=_snake_case						)




			@apply_forward_hook
			def   lowerCamelCase__  (					self    :       Optional[int]							, _snake_case    :       torch.FloatTensor							, _snake_case    :       bool = True						)   ->			Union[DecoderOutput, torch.FloatTensor]:







								"""simple docstring"""
								if self.use_slicing and z.shape[0] > 1:
													A_								=    [self._decode(_snake_case						).sample for z_slice in z.split(1						)]
													A_								=    torch.cat(_snake_case						)
								else:
													A_								=    self._decode(_snake_case						).sample

								if not return_dict:
													return (decoded,)

								return DecoderOutput(sample=_snake_case						)




			def   lowerCamelCase__  (					self    :       int							, _snake_case    :       Optional[int]							, _snake_case    :       Tuple							, _snake_case    :       Optional[int]						)   ->			Optional[int]:







								"""simple docstring"""
								A_								=    min(a.shape[2]							, b.shape[2]							, _snake_case						)
								for y in range(_snake_case						):
													A_								=    a[:, :, -blend_extent + y, :] * (1 - y / blend_extent) + b[:, :, y, :] * (y / blend_extent)
								return b




			def   lowerCamelCase__  (					self    :       Any							, _snake_case    :       Dict							, _snake_case    :       Optional[Any]							, _snake_case    :       Dict						)   ->			Union[str, Any]:







								"""simple docstring"""
								A_								=    min(a.shape[3]							, b.shape[3]							, _snake_case						)
								for x in range(_snake_case						):
													A_								=    a[:, :, :, -blend_extent + x] * (1 - x / blend_extent) + b[:, :, :, x] * (x / blend_extent)
								return b




			def   lowerCamelCase__  (					self    :       Optional[int]							, _snake_case    :       torch.FloatTensor							, _snake_case    :       bool = True						)   ->			AutoencoderKLOutput:







								"""simple docstring"""
								A_								=    int(self.tile_sample_min_size * (1 - self.tile_overlap_factor)						)
								A_								=    int(self.tile_latent_min_size * self.tile_overlap_factor						)
								A_								=    self.tile_latent_min_size - blend_extent

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

								A_								=    torch.cat(_snake_case							, dim=2						)
								A_								=    DiagonalGaussianDistribution(_snake_case						)

								if not return_dict:
													return (posterior,)

								return AutoencoderKLOutput(latent_dist=_snake_case						)




			def   lowerCamelCase__  (					self    :       Dict							, _snake_case    :       torch.FloatTensor							, _snake_case    :       bool = True						)   ->			Union[DecoderOutput, torch.FloatTensor]:







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

								A_								=    torch.cat(_snake_case							, dim=2						)
								if not return_dict:
													return (dec,)

								return DecoderOutput(sample=_snake_case						)






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







								"""simple docstring"""
								A_								=    sample
								A_								=    self.encode(_snake_case						).latent_dist
								if sample_posterior:
													A_								=    posterior.sample(generator=_snake_case						)
								else:
													A_								=    posterior.mode()
								A_								=    self.decode(_snake_case						).sample

								if not return_dict:
													return (dec,)

								return DecoderOutput(sample=_snake_case						)






"""simple docstring"""





from ..utils import is_flax_available, is_torch_available


if is_torch_available():
		from .autoencoder_kl import AutoencoderKL
		from .controlnet import ControlNetModel
		from .dual_transformer_ad import DualTransformeraDModel
		from .modeling_utils import ModelMixin
		from .prior_transformer import PriorTransformer
		from .ta_film_transformer import TaFilmDecoder
		from .transformer_ad import TransformeraDModel
		from .unet_ad import UNetaDModel
		from .unet_ad import UNetaDModel
		from .unet_ad_condition import UNetaDConditionModel
		from .unet_ad_condition import UNetaDConditionModel
		from .vq_model import VQModel

if is_flax_available():
		from .controlnet_flax import FlaxControlNetModel
		from .unet_ad_condition_flax import FlaxUNetaDConditionModel
		from .vae_flax import FlaxAutoencoderKL




