






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 (				snake_case__ ):







		"""simple docstring"""



		snake_case    =   42


class      UpperCamelCase (				snake_case__			,       snake_case__ ):







		"""simple docstring"""



		snake_case    =   True
		@register_to_config
		def __init__(  self		:       Union[str, Any]						,_SCREAMING_SNAKE_CASE		:       int = 3						,_SCREAMING_SNAKE_CASE		:       int = 3						,_SCREAMING_SNAKE_CASE		:       Tuple[str] = ("DownEncoderBlock2D",)						,_SCREAMING_SNAKE_CASE		:       Tuple[str] = ("UpDecoderBlock2D",)						,_SCREAMING_SNAKE_CASE		:       Tuple[int] = (6_4,)						,_SCREAMING_SNAKE_CASE		:       int = 1						,_SCREAMING_SNAKE_CASE		:       str = "silu"						,_SCREAMING_SNAKE_CASE		:       int = 4						,_SCREAMING_SNAKE_CASE		:       int = 3_2						,_SCREAMING_SNAKE_CASE		:       int = 3_2						,_SCREAMING_SNAKE_CASE		:       float = 0.18215						,)    ->   List[str]:


						'''simple docstring'''
						super().__init__()

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

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

						A	 =      nn.Convad(2 * latent_channels						,2 * latent_channels						,1		)
						A	 =      nn.Convad(_SCREAMING_SNAKE_CASE						,_SCREAMING_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.25
		def       A(  self		:       Optional[Any]						,_SCREAMING_SNAKE_CASE		:       Optional[int]						,_SCREAMING_SNAKE_CASE		:       Optional[Any]=False		)    ->   int:


						'''simple docstring'''
						if isinstance(_SCREAMING_SNAKE_CASE						,(Encoder, Decoder)		):
										A	 =      value
		def       A(  self		:       List[str]						,_SCREAMING_SNAKE_CASE		:       bool = True		)    ->   Optional[Any]:


						'''simple docstring'''
						A	 =      use_tiling
		def       A(  self		:       Any		)    ->   List[str]:


						'''simple docstring'''
						self.enable_tiling(_SCREAMING_SNAKE_CASE		)
		def       A(  self		:       Optional[Any]		)    ->   Optional[int]:


						'''simple docstring'''
						A	 =      True
		def       A(  self		:       Tuple		)    ->   int:


						'''simple docstring'''
						A	 =      False
		@property
		# Copied from diffusers.models.unet_2d_condition.UNet2DConditionModel.attn_processors
		def       A(  self		:       Dict		)    ->   Dict[str, AttentionProcessor]:


						'''simple docstring'''
						A	 =      {}

						def fn_recursive_add_processors(_SCREAMING_SNAKE_CASE		:       str						,_SCREAMING_SNAKE_CASE		:       torch.nn.Module						,_SCREAMING_SNAKE_CASE		:       Dict[str, AttentionProcessor]		):
										if hasattr(_SCREAMING_SNAKE_CASE						,'set_processor'		):
														A	 =      module.processor

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

										return processors

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

						return processors
		def       A(  self		:       str						,_SCREAMING_SNAKE_CASE		:       Union[AttentionProcessor, Dict[str, AttentionProcessor]]		)    ->   Any:


						'''simple docstring'''
						A	 =      len(self.attn_processors.keys()		)

						if isinstance(_SCREAMING_SNAKE_CASE						,_SCREAMING_SNAKE_CASE		) and len(_SCREAMING_SNAKE_CASE		) != count:
										raise ValueError(
										    f'A dict of processors was passed, but the number of processors {len(_SCREAMING_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(_SCREAMING_SNAKE_CASE		:       str						,_SCREAMING_SNAKE_CASE		:       torch.nn.Module						,_SCREAMING_SNAKE_CASE		:       Optional[Any]		):
										if hasattr(_SCREAMING_SNAKE_CASE						,'set_processor'		):
														if not isinstance(_SCREAMING_SNAKE_CASE						,_SCREAMING_SNAKE_CASE		):
																		module.set_processor(_SCREAMING_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}'						,_SCREAMING_SNAKE_CASE						,_SCREAMING_SNAKE_CASE		)

						for name, module in self.named_children():
										fn_recursive_attn_processor(_SCREAMING_SNAKE_CASE						,_SCREAMING_SNAKE_CASE						,_SCREAMING_SNAKE_CASE		)
		def       A(  self		:       Union[str, Any]		)    ->   List[Any]:


						'''simple docstring'''
						self.set_attn_processor(AttnProcessor()		)
		@apply_forward_hook
		def       A(  self		:       Optional[int]						,_SCREAMING_SNAKE_CASE		:       torch.FloatTensor						,_SCREAMING_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(_SCREAMING_SNAKE_CASE						,return_dict=_SCREAMING_SNAKE_CASE		)

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

						A	 =      self.quant_conv(_SCREAMING_SNAKE_CASE		)
						A	 =      DiagonalGaussianDistribution(_SCREAMING_SNAKE_CASE		)

						if not return_dict:
										return (posterior,)

						return AutoencoderKLOutput(latent_dist=_SCREAMING_SNAKE_CASE		)
		def       A(  self		:       int						,_SCREAMING_SNAKE_CASE		:       torch.FloatTensor						,_SCREAMING_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(_SCREAMING_SNAKE_CASE						,return_dict=_SCREAMING_SNAKE_CASE		)

						A	 =      self.post_quant_conv(_SCREAMING_SNAKE_CASE		)
						A	 =      self.decoder(_SCREAMING_SNAKE_CASE		)

						if not return_dict:
										return (dec,)

						return DecoderOutput(sample=_SCREAMING_SNAKE_CASE		)
		@apply_forward_hook
		def       A(  self		:       Tuple						,_SCREAMING_SNAKE_CASE		:       torch.FloatTensor						,_SCREAMING_SNAKE_CASE		:       bool = True		)    ->   Union[DecoderOutput, torch.FloatTensor]:


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

						if not return_dict:
										return (decoded,)

						return DecoderOutput(sample=_SCREAMING_SNAKE_CASE		)
		def       A(  self		:       Optional[Any]						,_SCREAMING_SNAKE_CASE		:       Dict						,_SCREAMING_SNAKE_CASE		:       List[str]						,_SCREAMING_SNAKE_CASE		:       Union[str, Any]		)    ->   Union[str, Any]:


						'''simple docstring'''
						A	 =      min(a.shape[2]						,b.shape[2]						,_SCREAMING_SNAKE_CASE		)
						for y in range(_SCREAMING_SNAKE_CASE		):
										A	 =      a[:, :, -blend_extent + y, :] * (1 - y / blend_extent) + b[:, :, y, :] * (y / blend_extent)
						return b
		def       A(  self		:       Optional[int]						,_SCREAMING_SNAKE_CASE		:       List[Any]						,_SCREAMING_SNAKE_CASE		:       int						,_SCREAMING_SNAKE_CASE		:       Optional[Any]		)    ->   Optional[int]:


						'''simple docstring'''
						A	 =      min(a.shape[3]						,b.shape[3]						,_SCREAMING_SNAKE_CASE		)
						for x in range(_SCREAMING_SNAKE_CASE		):
										A	 =      a[:, :, :, -blend_extent + x] * (1 - x / blend_extent) + b[:, :, :, x] * (x / blend_extent)
						return b
		def       A(  self		:       Optional[int]						,_SCREAMING_SNAKE_CASE		:       torch.FloatTensor						,_SCREAMING_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]						,_SCREAMING_SNAKE_CASE		):
										A	 =      []
										for j in range(0						,x.shape[3]						,_SCREAMING_SNAKE_CASE		):
														A	 =      x[:, :, i : i + self.tile_sample_min_size, j : j + self.tile_sample_min_size]
														A	 =      self.encoder(_SCREAMING_SNAKE_CASE		)
														A	 =      self.quant_conv(_SCREAMING_SNAKE_CASE		)
														row.append(_SCREAMING_SNAKE_CASE		)
										rows.append(_SCREAMING_SNAKE_CASE		)
						A	 =      []
						for i, row in enumerate(_SCREAMING_SNAKE_CASE		):
										A	 =      []
										for j, tile in enumerate(_SCREAMING_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]						,_SCREAMING_SNAKE_CASE						,_SCREAMING_SNAKE_CASE		)
														if j > 0:
																		A	 =      self.blend_h(row[j - 1]						,_SCREAMING_SNAKE_CASE						,_SCREAMING_SNAKE_CASE		)
														result_row.append(tile[:, :, :row_limit, :row_limit]		)
										result_rows.append(torch.cat(_SCREAMING_SNAKE_CASE						,dim=3		)		)

						A	 =      torch.cat(_SCREAMING_SNAKE_CASE						,dim=2		)
						A	 =      DiagonalGaussianDistribution(_SCREAMING_SNAKE_CASE		)

						if not return_dict:
										return (posterior,)

						return AutoencoderKLOutput(latent_dist=_SCREAMING_SNAKE_CASE		)
		def       A(  self		:       Optional[Any]						,_SCREAMING_SNAKE_CASE		:       torch.FloatTensor						,_SCREAMING_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]						,_SCREAMING_SNAKE_CASE		):
										A	 =      []
										for j in range(0						,z.shape[3]						,_SCREAMING_SNAKE_CASE		):
														A	 =      z[:, :, i : i + self.tile_latent_min_size, j : j + self.tile_latent_min_size]
														A	 =      self.post_quant_conv(_SCREAMING_SNAKE_CASE		)
														A	 =      self.decoder(_SCREAMING_SNAKE_CASE		)
														row.append(_SCREAMING_SNAKE_CASE		)
										rows.append(_SCREAMING_SNAKE_CASE		)
						A	 =      []
						for i, row in enumerate(_SCREAMING_SNAKE_CASE		):
										A	 =      []
										for j, tile in enumerate(_SCREAMING_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]						,_SCREAMING_SNAKE_CASE						,_SCREAMING_SNAKE_CASE		)
														if j > 0:
																		A	 =      self.blend_h(row[j - 1]						,_SCREAMING_SNAKE_CASE						,_SCREAMING_SNAKE_CASE		)
														result_row.append(tile[:, :, :row_limit, :row_limit]		)
										result_rows.append(torch.cat(_SCREAMING_SNAKE_CASE						,dim=3		)		)

						A	 =      torch.cat(_SCREAMING_SNAKE_CASE						,dim=2		)
						if not return_dict:
										return (dec,)

						return DecoderOutput(sample=_SCREAMING_SNAKE_CASE		)







		def       A(  self		:       Tuple						,_SCREAMING_SNAKE_CASE		:       torch.FloatTensor						,_SCREAMING_SNAKE_CASE		:       bool = False						,_SCREAMING_SNAKE_CASE		:       bool = True						,_SCREAMING_SNAKE_CASE		:       Optional[torch.Generator] = None						,)    ->   Union[DecoderOutput, torch.FloatTensor]:


						'''simple docstring'''
						A	 =      sample
						A	 =      self.encode(_SCREAMING_SNAKE_CASE		).latent_dist
						if sample_posterior:
										A	 =      posterior.sample(generator=_SCREAMING_SNAKE_CASE		)
						else:
										A	 =      posterior.mode()
						A	 =      self.decode(_SCREAMING_SNAKE_CASE		).sample

						if not return_dict:
										return (dec,)

						return DecoderOutput(sample=_SCREAMING_SNAKE_CASE		)













import os


# All paths are set with the intent you should run this script from the root of the repo with the command
# python utils/check_doctest_list.py
lowerCAmelCase_           = '.'


if __name__ == "__main__":
					lowerCAmelCase_           = os.path.join(REPO_PATH, 'utils/documentation_tests.txt')
					lowerCAmelCase_           = []
					lowerCAmelCase_           = []
					with open(doctest_file_path) as fp:
										for line in fp:
															lowerCAmelCase_           = line.strip()
															lowerCAmelCase_           = os.path.join(REPO_PATH, line)
															if not (os.path.isfile(path) or os.path.isdir(path)):
																				non_existent_paths.append(line)
															all_paths.append(path)
					if len(non_existent_paths) > 0:
										lowerCAmelCase_           = '\n'.join(non_existent_paths)
										raise ValueError(f'''`utils/documentation_tests.txt` contains non-existent paths:\n{non_existent_paths}''')
					if all_paths != sorted(all_paths):
										raise ValueError('Files in `utils/documentation_tests.txt` are not in alphabetical order.')





