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  SCREAMING_SNAKE_CASE_ (					lowerCAmelCase_			):
  '''simple docstring'''


  lowercase		:     "DiagonalGaussianDistribution"







class  SCREAMING_SNAKE_CASE_ (					lowerCAmelCase_   ,						lowerCAmelCase_			):
  '''simple docstring'''


  lowercase		:     str            =				True







  @register_to_config
  def __init__(				self    :					str      ,	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] = (64,)      ,	SCREAMING_SNAKE_CASE__    :					int = 1      ,	SCREAMING_SNAKE_CASE__    :					str = "silu"      ,	SCREAMING_SNAKE_CASE__    :					int = 4      ,	SCREAMING_SNAKE_CASE__    :					int = 32      ,	SCREAMING_SNAKE_CASE__    :					int = 32      ,	SCREAMING_SNAKE_CASE__    :					float = 0.1_8_2_1_5      ,	)    ->				Optional[Any]:
         super().__init__()

         # pass init params to Encoder
         A						:     Any         =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						:     Any         =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						:     Union[str, Any]         =nn.Convad(2 * latent_channels      ,	2 * latent_channels      ,	1     )
         A						:     Tuple         =nn.Convad(SCREAMING_SNAKE_CASE__      ,	SCREAMING_SNAKE_CASE__      ,	1     )

         A						:     List[str]         =False
         A						:     Optional[Any]         =False

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







  def 		SCREAMING_SNAKE_CASE_		(				self    :					List[str]      ,	SCREAMING_SNAKE_CASE__    :					Optional[Any]      ,	SCREAMING_SNAKE_CASE__    :					List[Any]=False     )    ->				Any:
         if isinstance(SCREAMING_SNAKE_CASE__      ,	(Encoder, Decoder)     ):
                A						:     Dict         =value







  def 		SCREAMING_SNAKE_CASE_		(				self    :					Optional[int]      ,	SCREAMING_SNAKE_CASE__    :					bool = True     )    ->				Any:
         A						:     Optional[Any]         =use_tiling







  def 		SCREAMING_SNAKE_CASE_		(				self    :					Tuple     )    ->				Tuple:
         self.enable_tiling(SCREAMING_SNAKE_CASE__     )







  def 		SCREAMING_SNAKE_CASE_		(				self    :					Dict     )    ->				int:
         A						:     Optional[Any]         =True







  def 		SCREAMING_SNAKE_CASE_		(				self    :					List[str]     )    ->				List[Any]:
         A						:     List[Any]         =False







  @property
  # Copied from diffusers.models.unet_2d_condition.UNet2DConditionModel.attn_processors
  def 		SCREAMING_SNAKE_CASE_		(				self    :					Union[str, Any]     )    ->				Dict[str, AttentionProcessor]:
         A						:     Optional[Any]         ={}

         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						:     Any         =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 		SCREAMING_SNAKE_CASE_		(				self    :					Optional[int]      ,	SCREAMING_SNAKE_CASE__    :					Union[AttentionProcessor, Dict[str, AttentionProcessor]]     )    ->				List[Any]:
         A						:     Optional[Any]         =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__    :					List[str]     ):
                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 		SCREAMING_SNAKE_CASE_		(				self    :					int     )    ->				int:
         self.set_attn_processor(AttnProcessor()     )







  @apply_forward_hook
  def 		SCREAMING_SNAKE_CASE_		(				self    :					Tuple      ,	SCREAMING_SNAKE_CASE__    :					torch.FloatTensor      ,	SCREAMING_SNAKE_CASE__    :					bool = True     )    ->				AutoencoderKLOutput:
         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						:     Union[str, Any]         =[self.encoder(SCREAMING_SNAKE_CASE__     ) for x_slice in x.split(1     )]
                A						:     Dict         =torch.cat(SCREAMING_SNAKE_CASE__     )
         else:
                A						:     Union[str, Any]         =self.encoder(SCREAMING_SNAKE_CASE__     )

         A						:     Tuple         =self.quant_conv(SCREAMING_SNAKE_CASE__     )
         A						:     List[Any]         =DiagonalGaussianDistribution(SCREAMING_SNAKE_CASE__     )

         if not return_dict:
                return (posterior,)

         return AutoencoderKLOutput(latent_dist=SCREAMING_SNAKE_CASE__     )







  def 		SCREAMING_SNAKE_CASE_		(				self    :					Optional[int]      ,	SCREAMING_SNAKE_CASE__    :					torch.FloatTensor      ,	SCREAMING_SNAKE_CASE__    :					bool = True     )    ->				Union[DecoderOutput, torch.FloatTensor]:
         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						:     Union[str, Any]         =self.post_quant_conv(SCREAMING_SNAKE_CASE__     )
         A						:     List[str]         =self.decoder(SCREAMING_SNAKE_CASE__     )

         if not return_dict:
                return (dec,)

         return DecoderOutput(sample=SCREAMING_SNAKE_CASE__     )







  @apply_forward_hook
  def 		SCREAMING_SNAKE_CASE_		(				self    :					Dict      ,	SCREAMING_SNAKE_CASE__    :					torch.FloatTensor      ,	SCREAMING_SNAKE_CASE__    :					bool = True     )    ->				Union[DecoderOutput, torch.FloatTensor]:
         if self.use_slicing and z.shape[0] > 1:
                A						:     Tuple         =[self._decode(SCREAMING_SNAKE_CASE__     ).sample for z_slice in z.split(1     )]
                A						:     Dict         =torch.cat(SCREAMING_SNAKE_CASE__     )
         else:
                A						:     List[str]         =self._decode(SCREAMING_SNAKE_CASE__     ).sample

         if not return_dict:
                return (decoded,)

         return DecoderOutput(sample=SCREAMING_SNAKE_CASE__     )







  def 		SCREAMING_SNAKE_CASE_		(				self    :					Any      ,	SCREAMING_SNAKE_CASE__    :					Tuple      ,	SCREAMING_SNAKE_CASE__    :					str      ,	SCREAMING_SNAKE_CASE__    :					List[str]     )    ->				Union[str, Any]:
         A						:     str         =min(a.shape[2]      ,	b.shape[2]      ,	SCREAMING_SNAKE_CASE__     )
         for y in range(SCREAMING_SNAKE_CASE__     ):
                A						:     str         =a[:, :, -blend_extent + y, :] * (1 - y / blend_extent) + b[:, :, y, :] * (y / blend_extent)
         return b







  def 		SCREAMING_SNAKE_CASE_		(				self    :					Tuple      ,	SCREAMING_SNAKE_CASE__    :					Optional[Any]      ,	SCREAMING_SNAKE_CASE__    :					Any      ,	SCREAMING_SNAKE_CASE__    :					Tuple     )    ->				Tuple:
         A						:     List[Any]         =min(a.shape[3]      ,	b.shape[3]      ,	SCREAMING_SNAKE_CASE__     )
         for x in range(SCREAMING_SNAKE_CASE__     ):
                A						:     Optional[Any]         =a[:, :, :, -blend_extent + x] * (1 - x / blend_extent) + b[:, :, :, x] * (x / blend_extent)
         return b







  def 		SCREAMING_SNAKE_CASE_		(				self    :					Tuple      ,	SCREAMING_SNAKE_CASE__    :					torch.FloatTensor      ,	SCREAMING_SNAKE_CASE__    :					bool = True     )    ->				AutoencoderKLOutput:
         A						:     int         =int(self.tile_sample_min_size * (1 - self.tile_overlap_factor)     )
         A						:     Optional[Any]         =int(self.tile_latent_min_size * self.tile_overlap_factor     )
         A						:     Dict         =self.tile_latent_min_size - blend_extent

         # Split the image into 512x512 tiles and encode them separately.
         A						:     Dict         =[]
         for i in range(0      ,	x.shape[2]      ,	SCREAMING_SNAKE_CASE__     ):
                A						:     Any         =[]
                for j in range(0      ,	x.shape[3]      ,	SCREAMING_SNAKE_CASE__     ):
                       A						:     str         =x[:, :, i : i + self.tile_sample_min_size, j : j + self.tile_sample_min_size]
                       A						:     Any         =self.encoder(SCREAMING_SNAKE_CASE__     )
                       A						:     List[Any]         =self.quant_conv(SCREAMING_SNAKE_CASE__     )
                       row.append(SCREAMING_SNAKE_CASE__     )
                rows.append(SCREAMING_SNAKE_CASE__     )
         A						:     int         =[]
         for i, row in enumerate(SCREAMING_SNAKE_CASE__     ):
                A						:     Optional[Any]         =[]
                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						:     List[str]         =self.blend_v(rows[i - 1][j]      ,	SCREAMING_SNAKE_CASE__      ,	SCREAMING_SNAKE_CASE__     )
                       if j > 0:
                              A						:     Any         =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						:     List[str]         =torch.cat(SCREAMING_SNAKE_CASE__      ,	dim=2     )
         A						:     int         =DiagonalGaussianDistribution(SCREAMING_SNAKE_CASE__     )

         if not return_dict:
                return (posterior,)

         return AutoencoderKLOutput(latent_dist=SCREAMING_SNAKE_CASE__     )







  def 		SCREAMING_SNAKE_CASE_		(				self    :					Optional[int]      ,	SCREAMING_SNAKE_CASE__    :					torch.FloatTensor      ,	SCREAMING_SNAKE_CASE__    :					bool = True     )    ->				Union[DecoderOutput, torch.FloatTensor]:
         A						:     List[str]         =int(self.tile_latent_min_size * (1 - self.tile_overlap_factor)     )
         A						:     str         =int(self.tile_sample_min_size * self.tile_overlap_factor     )
         A						:     Tuple         =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						:     Optional[int]         =[]
         for i in range(0      ,	z.shape[2]      ,	SCREAMING_SNAKE_CASE__     ):
                A						:     Any         =[]
                for j in range(0      ,	z.shape[3]      ,	SCREAMING_SNAKE_CASE__     ):
                       A						:     Any         =z[:, :, i : i + self.tile_latent_min_size, j : j + self.tile_latent_min_size]
                       A						:     Dict         =self.post_quant_conv(SCREAMING_SNAKE_CASE__     )
                       A						:     str         =self.decoder(SCREAMING_SNAKE_CASE__     )
                       row.append(SCREAMING_SNAKE_CASE__     )
                rows.append(SCREAMING_SNAKE_CASE__     )
         A						:     List[Any]         =[]
         for i, row in enumerate(SCREAMING_SNAKE_CASE__     ):
                A						:     Optional[Any]         =[]
                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						:     Optional[int]         =self.blend_v(rows[i - 1][j]      ,	SCREAMING_SNAKE_CASE__      ,	SCREAMING_SNAKE_CASE__     )
                       if j > 0:
                              A						:     Optional[Any]         =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						:     Optional[Any]         =torch.cat(SCREAMING_SNAKE_CASE__      ,	dim=2     )
         if not return_dict:
                return (dec,)

         return DecoderOutput(sample=SCREAMING_SNAKE_CASE__     )







  def 		SCREAMING_SNAKE_CASE_		(				self    :					int      ,	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]:
         A						:     List[Any]         =sample
         A						:     str         =self.encode(SCREAMING_SNAKE_CASE__     ).latent_dist
         if sample_posterior:
                A						:     Tuple         =posterior.sample(generator=SCREAMING_SNAKE_CASE__     )
         else:
                A						:     Tuple         =posterior.mode()
         A						:     str         =self.decode(SCREAMING_SNAKE_CASE__     ).sample

         if not return_dict:
                return (dec,)

         return DecoderOutput(sample=SCREAMING_SNAKE_CASE__     )







from typing import List

from .keymap import KEYMAP, get_character



def A__  (			lowercase:     str					)				->		List[str]:

       def decorator(lowercase:     int					):
              A						:     Tuple         =getattr(lowercase,		'handle_key',		[]					)
              handle += [key]
              setattr(lowercase,		'handle_key',		lowercase					)
              return func

       return decorator



def A__  (			*lowercase:     List[str]					)				->		Dict:

       def decorator(lowercase:     Union[str, Any]					):
              A						:     Optional[int]         =getattr(lowercase,		'handle_key',		[]					)
              handle += keys
              setattr(lowercase,		'handle_key',		lowercase					)
              return func

       return decorator



class  SCREAMING_SNAKE_CASE_ (					lowerCAmelCase_			):
  '''simple docstring'''


  def __new__(				cls    :					List[str]      ,	SCREAMING_SNAKE_CASE__    :					Optional[Any]      ,	SCREAMING_SNAKE_CASE__    :					List[Any]      ,	SCREAMING_SNAKE_CASE__    :					List[str]     )    ->				Any:
         A						:     Dict         =super().__new__(cls      ,	SCREAMING_SNAKE_CASE__      ,	SCREAMING_SNAKE_CASE__      ,	SCREAMING_SNAKE_CASE__     )
         if not hasattr(SCREAMING_SNAKE_CASE__      ,	'key_handler'     ):
                setattr(SCREAMING_SNAKE_CASE__      ,	'key_handler'      ,	{}     )
         setattr(SCREAMING_SNAKE_CASE__      ,	'handle_input'      ,	KeyHandler.handle_input     )

         for value in attrs.values():
                A						:     Optional[Any]         =getattr(SCREAMING_SNAKE_CASE__      ,	'handle_key'      ,	[]     )
                for key in handled_keys:
                       A						:     str         =value
         return new_cls







  @staticmethod
  def 		SCREAMING_SNAKE_CASE_		(				cls    :					str     )    ->				Any:
         A						:     str         =get_character()
         if char != KEYMAP["undefined"]:
                A						:     List[str]         =ord(SCREAMING_SNAKE_CASE__     )
         A						:     List[str]         =cls.key_handler.get(SCREAMING_SNAKE_CASE__     )
         if handler:
                A						:     List[str]         =char
                return handler(cls     )
         else:
                return None







def A__  (			cls:     Optional[int]					)				->		str:
       return KeyHandler(cls.__name__,		cls.__bases__,		cls.__dict__.copy()					)






