



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


 '''simple docstring'''




 __magic_name__					: "DiagonalGaussianDistribution"







class 	UpperCAmelCase	(		__a					,	__a):


 '''simple docstring'''




 __magic_name__					: List[str]  =      True




 @register_to_config
 def __init__(   self							,  lowerCAmelCase_ = 3							,  lowerCAmelCase_ = 3							,  lowerCAmelCase_ = ("DownEncoderBlock2D",)							,  lowerCAmelCase_ = ("UpDecoderBlock2D",)							,  lowerCAmelCase_ = (6_4,)							,  lowerCAmelCase_ = 1							,  lowerCAmelCase_ = "silu"							,  lowerCAmelCase_ = 4							,  lowerCAmelCase_ = 3_2							,  lowerCAmelCase_ = 3_2							,  lowerCAmelCase_ = 0.1_8_2_1_5							,  ) ->      str:



      """simple docstring"""







      super().__init__()

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

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

      a_       		=nn.Convad(2 * latent_channels							,  2 * latent_channels							,  1)
      a_       		=nn.Convad(lowerCAmelCase_							,  lowerCAmelCase_							,  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       lowercase_  (   self							,  lowerCAmelCase_							,  lowerCAmelCase_=False) ->      Tuple:



      """simple docstring"""







      if isinstance(lowerCAmelCase_							,  (Encoder, Decoder)):
           a_       		=value




 def       lowercase_  (   self							,  lowerCAmelCase_ = True) ->      Union[str, Any]:



      """simple docstring"""







      a_       		=use_tiling




 def       lowercase_  (   self) ->      str:



      """simple docstring"""







      self.enable_tiling(lowerCAmelCase_)




 def       lowercase_  (   self) ->      Tuple:



      """simple docstring"""







      a_       		=True




 def       lowercase_  (   self) ->      Dict:



      """simple docstring"""







      a_       		=False




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



      """simple docstring"""







      a_       		={}

      def fn_recursive_add_processors(lowerCAmelCase_							,  lowerCAmelCase_							,  lowerCAmelCase_):
           if hasattr(lowerCAmelCase_							,  "set_processor"):
                a_       		=module.processor

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

           return processors

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

      return processors




 def       lowercase_  (   self							,  lowerCAmelCase_) ->      int:



      """simple docstring"""







      a_       		=len(self.attn_processors.keys())

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

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

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




 def       lowercase_  (   self) ->      int:



      """simple docstring"""







      self.set_attn_processor(AttnProcessor())




 @apply_forward_hook
 def       lowercase_  (   self							,  lowerCAmelCase_							,  lowerCAmelCase_ = 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(lowerCAmelCase_							,  return_dict=lowerCAmelCase_)

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

      a_       		=self.quant_conv(lowerCAmelCase_)
      a_       		=DiagonalGaussianDistribution(lowerCAmelCase_)

      if not return_dict:
           return (posterior,)

      return AutoencoderKLOutput(latent_dist=lowerCAmelCase_)




 def       lowercase_  (   self							,  lowerCAmelCase_							,  lowerCAmelCase_ = 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(lowerCAmelCase_							,  return_dict=lowerCAmelCase_)

      a_       		=self.post_quant_conv(lowerCAmelCase_)
      a_       		=self.decoder(lowerCAmelCase_)

      if not return_dict:
           return (dec,)

      return DecoderOutput(sample=lowerCAmelCase_)




 @apply_forward_hook
 def       lowercase_  (   self							,  lowerCAmelCase_							,  lowerCAmelCase_ = True) ->      Union[DecoderOutput, torch.FloatTensor]:



      """simple docstring"""







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

      if not return_dict:
           return (decoded,)

      return DecoderOutput(sample=lowerCAmelCase_)




 def       lowercase_  (   self							,  lowerCAmelCase_							,  lowerCAmelCase_							,  lowerCAmelCase_) ->      int:



      """simple docstring"""







      a_       		=min(a.shape[2]							,  b.shape[2]							,  lowerCAmelCase_)
      for y in range(lowerCAmelCase_):
           a_       		=a[:, :, -blend_extent + y, :] * (1 - y / blend_extent) + b[:, :, y, :] * (y / blend_extent)
      return b




 def       lowercase_  (   self							,  lowerCAmelCase_							,  lowerCAmelCase_							,  lowerCAmelCase_) ->      Union[str, Any]:



      """simple docstring"""







      a_       		=min(a.shape[3]							,  b.shape[3]							,  lowerCAmelCase_)
      for x in range(lowerCAmelCase_):
           a_       		=a[:, :, :, -blend_extent + x] * (1 - x / blend_extent) + b[:, :, :, x] * (x / blend_extent)
      return b




 def       lowercase_  (   self							,  lowerCAmelCase_							,  lowerCAmelCase_ = 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]							,  lowerCAmelCase_):
           a_       		=[]
           for j in range(0							,  x.shape[3]							,  lowerCAmelCase_):
                a_       		=x[:, :, i : i + self.tile_sample_min_size, j : j + self.tile_sample_min_size]
                a_       		=self.encoder(lowerCAmelCase_)
                a_       		=self.quant_conv(lowerCAmelCase_)
                row.append(lowerCAmelCase_)
           rows.append(lowerCAmelCase_)
      a_       		=[]
      for i, row in enumerate(lowerCAmelCase_):
           a_       		=[]
           for j, tile in enumerate(lowerCAmelCase_):
                # 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]							,  lowerCAmelCase_							,  lowerCAmelCase_)
                if j > 0:
                     a_       		=self.blend_h(row[j - 1]							,  lowerCAmelCase_							,  lowerCAmelCase_)
                result_row.append(tile[:, :, :row_limit, :row_limit])
           result_rows.append(torch.cat(lowerCAmelCase_							,  dim=3))

      a_       		=torch.cat(lowerCAmelCase_							,  dim=2)
      a_       		=DiagonalGaussianDistribution(lowerCAmelCase_)

      if not return_dict:
           return (posterior,)

      return AutoencoderKLOutput(latent_dist=lowerCAmelCase_)




 def       lowercase_  (   self							,  lowerCAmelCase_							,  lowerCAmelCase_ = 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]							,  lowerCAmelCase_):
           a_       		=[]
           for j in range(0							,  z.shape[3]							,  lowerCAmelCase_):
                a_       		=z[:, :, i : i + self.tile_latent_min_size, j : j + self.tile_latent_min_size]
                a_       		=self.post_quant_conv(lowerCAmelCase_)
                a_       		=self.decoder(lowerCAmelCase_)
                row.append(lowerCAmelCase_)
           rows.append(lowerCAmelCase_)
      a_       		=[]
      for i, row in enumerate(lowerCAmelCase_):
           a_       		=[]
           for j, tile in enumerate(lowerCAmelCase_):
                # 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]							,  lowerCAmelCase_							,  lowerCAmelCase_)
                if j > 0:
                     a_       		=self.blend_h(row[j - 1]							,  lowerCAmelCase_							,  lowerCAmelCase_)
                result_row.append(tile[:, :, :row_limit, :row_limit])
           result_rows.append(torch.cat(lowerCAmelCase_							,  dim=3))

      a_       		=torch.cat(lowerCAmelCase_							,  dim=2)
      if not return_dict:
           return (dec,)

      return DecoderOutput(sample=lowerCAmelCase_)




 def       lowercase_  (   self							,  lowerCAmelCase_							,  lowerCAmelCase_ = False							,  lowerCAmelCase_ = True							,  lowerCAmelCase_ = None							,  ) ->      Union[DecoderOutput, torch.FloatTensor]:



      """simple docstring"""







      a_       		=sample
      a_       		=self.encode(lowerCAmelCase_).latent_dist
      if sample_posterior:
           a_       		=posterior.sample(generator=lowerCAmelCase_)
      else:
           a_       		=posterior.mode()
      a_       		=self.decode(lowerCAmelCase_).sample

      if not return_dict:
           return (dec,)

      return DecoderOutput(sample=lowerCAmelCase_)







'''simple docstring'''


import unittest
from typing import Tuple

import torch

from diffusers.utils import floats_tensor, randn_tensor, torch_all_close, torch_device
from diffusers.utils.testing_utils import require_torch







@require_torch
class 	UpperCAmelCase	:


 '''simple docstring'''




 @property
 def       lowercase_  (   self) ->      Any:



      """simple docstring"""







      return self.get_dummy_input()




 @property
 def       lowercase_  (   self) ->      List[str]:



      """simple docstring"""







      if self.block_type == "down":
           return (4, 3_2, 1_6, 1_6)
      elif self.block_type == "mid":
           return (4, 3_2, 3_2, 3_2)
      elif self.block_type == "up":
           return (4, 3_2, 6_4, 6_4)

      raise ValueError(f"""'{self.block_type}' is not a supported block_type. Set it to 'up', 'mid', or 'down'.""")




 def       lowercase_  (   self							,  lowerCAmelCase_=True							,  lowerCAmelCase_=False							,  lowerCAmelCase_=False							,  lowerCAmelCase_=False							,  ) ->      Dict:



      """simple docstring"""







      a_       		=4
      a_       		=3_2
      a_       		=(3_2, 3_2)

      a_       		=torch.manual_seed(0)
      a_       		=torch.device(lowerCAmelCase_)
      a_       		=(batch_size, num_channels) + sizes
      a_       		=randn_tensor(lowerCAmelCase_							,  generator=lowerCAmelCase_							,  device=lowerCAmelCase_)
      a_       		={"hidden_states": hidden_states}

      if include_temb:
           a_       		=1_2_8
           a_       		=randn_tensor((batch_size, temb_channels)							,  generator=lowerCAmelCase_							,  device=lowerCAmelCase_)

      if include_res_hidden_states_tuple:
           a_       		=torch.manual_seed(1)
           a_       		=(randn_tensor(lowerCAmelCase_							,  generator=lowerCAmelCase_							,  device=lowerCAmelCase_),)

      if include_encoder_hidden_states:
           a_       		=floats_tensor((batch_size, 3_2, 3_2)).to(lowerCAmelCase_)

      if include_skip_sample:
           a_       		=randn_tensor(((batch_size, 3) + sizes)							,  generator=lowerCAmelCase_							,  device=lowerCAmelCase_)

      return dummy_input




 def       lowercase_  (   self) ->      Optional[int]:



      """simple docstring"""







      a_       		={
          "in_channels": 3_2,
          "out_channels": 3_2,
          "temb_channels": 1_2_8,
      }
      if self.block_type == "up":
           a_       		=3_2

      if self.block_type == "mid":
           init_dict.pop("out_channels")

      a_       		=self.dummy_input
      return init_dict, inputs_dict




 def       lowercase_  (   self							,  lowerCAmelCase_) ->      Dict:



      """simple docstring"""







      a_		,						a_       		=self.prepare_init_args_and_inputs_for_common()
      a_       		=self.block_class(**lowerCAmelCase_)
      unet_block.to(lowerCAmelCase_)
      unet_block.eval()

      with torch.no_grad():
           a_       		=unet_block(**lowerCAmelCase_)

      if isinstance(lowerCAmelCase_							,  lowerCAmelCase_):
           a_       		=output[0]

      self.assertEqual(output.shape							,  self.output_shape)

      a_       		=output[0, -1, -3:, -3:]
      a_       		=torch.tensor(lowerCAmelCase_).to(lowerCAmelCase_)
      assert torch_all_close(output_slice.flatten()							,  lowerCAmelCase_							,  atol=5e-3)




 @unittest.skipIf(torch_device == "mps"							,  "Training is not supported in mps")
 def       lowercase_  (   self) ->      Optional[Any]:



      """simple docstring"""







      a_		,						a_       		=self.prepare_init_args_and_inputs_for_common()
      a_       		=self.block_class(**lowerCAmelCase_)
      model.to(lowerCAmelCase_)
      model.train()
      a_       		=model(**lowerCAmelCase_)

      if isinstance(lowerCAmelCase_							,  lowerCAmelCase_):
           a_       		=output[0]

      a_       		=torch.device(lowerCAmelCase_)
      a_       		=randn_tensor(output.shape							,  device=lowerCAmelCase_)
      a_       		=torch.nn.functional.mse_loss(lowerCAmelCase_							,  lowerCAmelCase_)
      loss.backward()


