




from ..utils import DummyObject, requires_backends

class   __a(					metaclass=_a       ):






    """simple docstring"""





    lowerCAmelCase        =				['''torch''', '''scipy''']





    def __init__(				self       ,*_SCREAMING_SNAKE_CASE       ,**_SCREAMING_SNAKE_CASE  )			->		Any:
         requires_backends(self       ,['''torch''', '''scipy''']  )





    @classmethod
    def 							a__				(				cls       ,*_SCREAMING_SNAKE_CASE       ,**_SCREAMING_SNAKE_CASE  )			->		List[Any]:
         requires_backends(cls       ,['''torch''', '''scipy''']  )





    @classmethod
    def 							a__				(				cls       ,*_SCREAMING_SNAKE_CASE       ,**_SCREAMING_SNAKE_CASE  )			->		str:
         requires_backends(cls       ,['''torch''', '''scipy''']  )





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   __a(					_a       ):






    """simple docstring"""





    lowerCAmelCase        =				42


class   __a(					_a    ,						_a       ):






    """simple docstring"""





    lowerCAmelCase        =				True





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

         # pass init params to Encoder
         UpperCAmelCase_      :							List[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
         UpperCAmelCase_      :							List[str]       =      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       ,)

         UpperCAmelCase_      :							int       =      nn.Convad(2 * latent_channels       ,2 * latent_channels       ,1  )
         UpperCAmelCase_      :							Union[str, Any]       =      nn.Convad(_SCREAMING_SNAKE_CASE       ,_SCREAMING_SNAKE_CASE       ,1  )

         UpperCAmelCase_      :							Optional[Any]       =      False
         UpperCAmelCase_      :							Dict       =      False

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





    def 							a__				(				self       ,_SCREAMING_SNAKE_CASE       ,_SCREAMING_SNAKE_CASE=False  )			->		int:
         if isinstance(_SCREAMING_SNAKE_CASE       ,(Encoder, Decoder)  ):
              UpperCAmelCase_      :							Union[str, Any]       =      value





    def 							a__				(				self       ,_SCREAMING_SNAKE_CASE = True  )			->		Optional[Any]:
         UpperCAmelCase_      :							Dict       =      use_tiling





    def 							a__				(				self  )			->		Optional[Any]:
         self.enable_tiling(_SCREAMING_SNAKE_CASE  )





    def 							a__				(				self  )			->		List[str]:
         UpperCAmelCase_      :							str       =      True





    def 							a__				(				self  )			->		Any:
         UpperCAmelCase_      :							Any       =      False





    @property
    # Copied from diffusers.models.unet_2d_condition.UNet2DConditionModel.attn_processors
    def 							a__				(				self  )			->		Dict[str, AttentionProcessor]:
         UpperCAmelCase_      :							int       =      {}

         def fn_recursive_add_processors(_SCREAMING_SNAKE_CASE       ,_SCREAMING_SNAKE_CASE       ,_SCREAMING_SNAKE_CASE  ):
              if hasattr(_SCREAMING_SNAKE_CASE       ,'''set_processor'''  ):
                   UpperCAmelCase_      :							Tuple       =      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       ,_SCREAMING_SNAKE_CASE  )			->		Union[str, Any]:
         UpperCAmelCase_      :							Optional[int]       =      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       ,_SCREAMING_SNAKE_CASE       ,_SCREAMING_SNAKE_CASE  ):
              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  )			->		Tuple:
         self.set_attn_processor(AttnProcessor()  )





    @apply_forward_hook
    def 							a__				(				self       ,_SCREAMING_SNAKE_CASE       ,_SCREAMING_SNAKE_CASE = 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:
              UpperCAmelCase_      :							int       =      [self.encoder(_SCREAMING_SNAKE_CASE  ) for x_slice in x.split(1  )]
              UpperCAmelCase_      :							List[str]       =      torch.cat(_SCREAMING_SNAKE_CASE  )
         else:
              UpperCAmelCase_      :							Optional[Any]       =      self.encoder(_SCREAMING_SNAKE_CASE  )

         UpperCAmelCase_      :							Any       =      self.quant_conv(_SCREAMING_SNAKE_CASE  )
         UpperCAmelCase_      :							Union[str, Any]       =      DiagonalGaussianDistribution(_SCREAMING_SNAKE_CASE  )

         if not return_dict:
              return (posterior,)

         return AutoencoderKLOutput(latent_dist=_SCREAMING_SNAKE_CASE  )





    def 							a__				(				self       ,_SCREAMING_SNAKE_CASE       ,_SCREAMING_SNAKE_CASE = 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  )

         UpperCAmelCase_      :							Tuple       =      self.post_quant_conv(_SCREAMING_SNAKE_CASE  )
         UpperCAmelCase_      :							Tuple       =      self.decoder(_SCREAMING_SNAKE_CASE  )

         if not return_dict:
              return (dec,)

         return DecoderOutput(sample=_SCREAMING_SNAKE_CASE  )





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

         if not return_dict:
              return (decoded,)

         return DecoderOutput(sample=_SCREAMING_SNAKE_CASE  )





    def 							a__				(				self       ,_SCREAMING_SNAKE_CASE       ,_SCREAMING_SNAKE_CASE       ,_SCREAMING_SNAKE_CASE  )			->		Union[str, Any]:
         UpperCAmelCase_      :							str       =      min(a.shape[2]       ,b.shape[2]       ,_SCREAMING_SNAKE_CASE  )
         for y in range(_SCREAMING_SNAKE_CASE  ):
              UpperCAmelCase_      :							str       =      a[:, :, -blend_extent + y, :] * (1 - y / blend_extent) + b[:, :, y, :] * (y / blend_extent)
         return b





    def 							a__				(				self       ,_SCREAMING_SNAKE_CASE       ,_SCREAMING_SNAKE_CASE       ,_SCREAMING_SNAKE_CASE  )			->		int:
         UpperCAmelCase_      :							Any       =      min(a.shape[3]       ,b.shape[3]       ,_SCREAMING_SNAKE_CASE  )
         for x in range(_SCREAMING_SNAKE_CASE  ):
              UpperCAmelCase_      :							List[str]       =      a[:, :, :, -blend_extent + x] * (1 - x / blend_extent) + b[:, :, :, x] * (x / blend_extent)
         return b





    def 							a__				(				self       ,_SCREAMING_SNAKE_CASE       ,_SCREAMING_SNAKE_CASE = True  )			->		AutoencoderKLOutput:
         UpperCAmelCase_      :							Optional[int]       =      int(self.tile_sample_min_size * (1 - self.tile_overlap_factor)  )
         UpperCAmelCase_      :							Any       =      int(self.tile_latent_min_size * self.tile_overlap_factor  )
         UpperCAmelCase_      :							str       =      self.tile_latent_min_size - blend_extent

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

         UpperCAmelCase_      :							Optional[Any]       =      torch.cat(_SCREAMING_SNAKE_CASE       ,dim=2  )
         UpperCAmelCase_      :							Union[str, Any]       =      DiagonalGaussianDistribution(_SCREAMING_SNAKE_CASE  )

         if not return_dict:
              return (posterior,)

         return AutoencoderKLOutput(latent_dist=_SCREAMING_SNAKE_CASE  )





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

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

         return DecoderOutput(sample=_SCREAMING_SNAKE_CASE  )





    def 							a__				(				self       ,_SCREAMING_SNAKE_CASE       ,_SCREAMING_SNAKE_CASE = False       ,_SCREAMING_SNAKE_CASE = True       ,_SCREAMING_SNAKE_CASE = None       ,)			->		Union[DecoderOutput, torch.FloatTensor]:
         UpperCAmelCase_      :							str       =      sample
         UpperCAmelCase_      :							Optional[Any]       =      self.encode(_SCREAMING_SNAKE_CASE  ).latent_dist
         if sample_posterior:
              UpperCAmelCase_      :							List[Any]       =      posterior.sample(generator=_SCREAMING_SNAKE_CASE  )
         else:
              UpperCAmelCase_      :							Union[str, Any]       =      posterior.mode()
         UpperCAmelCase_      :							List[Any]       =      self.decode(_SCREAMING_SNAKE_CASE  ).sample

         if not return_dict:
              return (dec,)

         return DecoderOutput(sample=_SCREAMING_SNAKE_CASE  )