





"""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 		lowercase					(					_UpperCAmelCase					):
      _SCREAMING_SNAKE_CASE					=      42







class 		lowercase					(					_UpperCAmelCase     ,  _UpperCAmelCase					):
      _SCREAMING_SNAKE_CASE					=      True





      @register_to_config
      def __init__(       self      ,						lowercase = 3      ,						lowercase = 3      ,						lowercase = ("DownEncoderBlock2D",)      ,						lowercase = ("UpDecoderBlock2D",)      ,						lowercase = (64,)      ,						lowercase = 1      ,						lowercase = "silu"      ,						lowercase = 4      ,						lowercase = 32      ,						lowercase = 32      ,						lowercase = 0.18_215      ,						)		->     Optional[int]:
            super().__init__()

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

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

            lowerCAmelCase  						=       nn.Convad(2 * latent_channels      ,						2 * latent_channels      ,						1 )
            lowerCAmelCase  						=       nn.Convad(__UpperCAmelCase      ,						__UpperCAmelCase      ,						1 )

            lowerCAmelCase  						=       False
            lowerCAmelCase  						=       False

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





      def     _snake_case			(       self      ,						lowercase      ,						lowercase=False )		->     Optional[Any]:

            if isinstance(__UpperCAmelCase      ,						(Encoder, Decoder) ):
                  lowerCAmelCase  						=       value





      def     _snake_case			(       self      ,						lowercase = True )		->     Tuple:
            lowerCAmelCase  						=       use_tiling





      def     _snake_case			(       self )		->     Tuple:
            self.enable_tiling(__UpperCAmelCase )





      def     _snake_case			(       self )		->     Union[str, Any]:
            lowerCAmelCase  						=       True





      def     _snake_case			(       self )		->     int:
            lowerCAmelCase  						=       False





      @property
      # Copied from diffusers.models.unet_2d_condition.UNet2DConditionModel.attn_processors
      def     _snake_case			(       self )		->     Tuple:
            lowerCAmelCase  						=       {}

            def fn_recursive_add_processors(lowercase      ,						lowercase      ,						lowercase ):
                  if hasattr(__UpperCAmelCase      ,						"""set_processor""" ):
                        lowerCAmelCase  						=       module.processor

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

                  return processors

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

            return processors





      def     _snake_case			(       self      ,						lowercase )		->     List[str]:
            lowerCAmelCase  						=       len(self.attn_processors.keys() )

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

            def fn_recursive_attn_processor(lowercase      ,						lowercase      ,						lowercase ):
                  if hasattr(__UpperCAmelCase      ,						"""set_processor""" ):
                        if not isinstance(__UpperCAmelCase      ,						__UpperCAmelCase ):
                              module.set_processor(__UpperCAmelCase )
                        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}'      ,						__UpperCAmelCase      ,						__UpperCAmelCase )

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





      def     _snake_case			(       self )		->     List[Any]:
            self.set_attn_processor(AttnProcessor() )





      @apply_forward_hook
      def     _snake_case			(       self      ,						lowercase      ,						lowercase = True )		->     List[str]:

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

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

            lowerCAmelCase  						=       self.quant_conv(__UpperCAmelCase )
            lowerCAmelCase  						=       DiagonalGaussianDistribution(__UpperCAmelCase )

            if not return_dict:
                  return (posterior,)

            return AutoencoderKLOutput(latent_dist=__UpperCAmelCase )





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

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

            lowerCAmelCase  						=       self.post_quant_conv(__UpperCAmelCase )
            lowerCAmelCase  						=       self.decoder(__UpperCAmelCase )

            if not return_dict:
                  return (dec,)

            return DecoderOutput(sample=__UpperCAmelCase )





      @apply_forward_hook
      def     _snake_case			(       self      ,						lowercase      ,						lowercase = True )		->     Optional[Any]:

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

            if not return_dict:
                  return (decoded,)

            return DecoderOutput(sample=__UpperCAmelCase )





      def     _snake_case			(       self      ,						lowercase      ,						lowercase      ,						lowercase )		->     List[Any]:
            lowerCAmelCase  						=       min(a.shape[2]      ,						b.shape[2]      ,						__UpperCAmelCase )
            for y in range(__UpperCAmelCase ):
                  lowerCAmelCase  						=       a[:, :, -blend_extent + y, :] * (1 - y / blend_extent) + b[:, :, y, :] * (y / blend_extent)
            return b





      def     _snake_case			(       self      ,						lowercase      ,						lowercase      ,						lowercase )		->     Tuple:
            lowerCAmelCase  						=       min(a.shape[3]      ,						b.shape[3]      ,						__UpperCAmelCase )
            for x in range(__UpperCAmelCase ):
                  lowerCAmelCase  						=       a[:, :, :, -blend_extent + x] * (1 - x / blend_extent) + b[:, :, :, x] * (x / blend_extent)
            return b





      def     _snake_case			(       self      ,						lowercase      ,						lowercase = True )		->     List[Any]:
            lowerCAmelCase  						=       int(self.tile_sample_min_size * (1 - self.tile_overlap_factor) )
            lowerCAmelCase  						=       int(self.tile_latent_min_size * self.tile_overlap_factor )
            lowerCAmelCase  						=       self.tile_latent_min_size - blend_extent

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

            lowerCAmelCase  						=       torch.cat(__UpperCAmelCase      ,						dim=2 )
            lowerCAmelCase  						=       DiagonalGaussianDistribution(__UpperCAmelCase )

            if not return_dict:
                  return (posterior,)

            return AutoencoderKLOutput(latent_dist=__UpperCAmelCase )





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

            lowerCAmelCase  						=       torch.cat(__UpperCAmelCase      ,						dim=2 )
            if not return_dict:
                  return (dec,)

            return DecoderOutput(sample=__UpperCAmelCase )





      def     _snake_case			(       self      ,						lowercase      ,						lowercase = False      ,						lowercase = True      ,						lowercase = None      ,						)		->     List[Any]:
            lowerCAmelCase  						=       sample
            lowerCAmelCase  						=       self.encode(__UpperCAmelCase ).latent_dist
            if sample_posterior:
                  lowerCAmelCase  						=       posterior.sample(generator=__UpperCAmelCase )
            else:
                  lowerCAmelCase  						=       posterior.mode()
            lowerCAmelCase  						=       self.decode(__UpperCAmelCase ).sample

            if not return_dict:
                  return (dec,)

            return DecoderOutput(sample=__UpperCAmelCase )













"""simple docstring"""





def  __A   (_SCREAMING_SNAKE_CASE     ,   _SCREAMING_SNAKE_CASE			)					->bool:





  """simple docstring"""


  return numa ^ numa < 0


if __name__ == "__main__":
  import doctest

  doctest.testmod()
