



"""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     a		(     lowerCAmelCase__				):
  UpperCamelCase		: "DiagonalGaussianDistribution"







class     a		(     lowerCAmelCase__		,					lowerCAmelCase__				):
  UpperCamelCase		: str     =    True



  @register_to_config
  def __init__(  self       :    Optional[int]				,						lowerCAmelCase       :    List[Any] = 3				,						lowerCAmelCase       :    Union[str, Any] = 3				,						lowerCAmelCase       :    Tuple = ("DownEncoderBlock2D",)				,						lowerCAmelCase       :    Optional[int] = ("UpDecoderBlock2D",)				,						lowerCAmelCase       :    List[Any] = (64,)				,						lowerCAmelCase       :    Optional[int] = 1				,						lowerCAmelCase       :    Tuple = "silu"				,						lowerCAmelCase       :    List[str] = 4				,						lowerCAmelCase       :    Optional[int] = 32				,						lowerCAmelCase       :    List[Any] = 32				,						lowerCAmelCase       :    List[Any] = 0.1_8_2_1_5				,						)   ->    str:


     '''simple docstring'''




     super().__init__()

     # pass init params to Encoder
     SCREAMING_SNAKE_CASE_:       Dict   	=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
     SCREAMING_SNAKE_CASE_:       Union[str, Any]   	=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				,						)

     SCREAMING_SNAKE_CASE_:       Dict   	=nn.Convad(2 * latent_channels				,						2 * latent_channels				,						1						)
     SCREAMING_SNAKE_CASE_:       Optional[int]   	=nn.Convad(__lowerCAmelCase				,						__lowerCAmelCase				,						1						)

     SCREAMING_SNAKE_CASE_:       List[str]   	=False
     SCREAMING_SNAKE_CASE_:       Tuple   	=False

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



  def 		lowerCamelCase__				(  self       :    Any				,						lowerCAmelCase       :    str				,						lowerCAmelCase       :    List[str]=False						)   ->    int:


     '''simple docstring'''




     if isinstance(__lowerCAmelCase				,						(Encoder, Decoder)						):
        SCREAMING_SNAKE_CASE_:       str   	=value



  def 		lowerCamelCase__				(  self       :    Dict				,						lowerCAmelCase       :    Optional[int] = True						)   ->    Optional[int]:


     '''simple docstring'''




     SCREAMING_SNAKE_CASE_:       List[Any]   	=use_tiling



  def 		lowerCamelCase__				(  self       :    Optional[Any]						)   ->    int:


     '''simple docstring'''




     self.enable_tiling(__lowerCAmelCase						)



  def 		lowerCamelCase__				(  self       :    int						)   ->    Tuple:


     '''simple docstring'''




     SCREAMING_SNAKE_CASE_:       str   	=True



  def 		lowerCamelCase__				(  self       :    int						)   ->    Dict:


     '''simple docstring'''




     SCREAMING_SNAKE_CASE_:       int   	=False



  @property
  # Copied from diffusers.models.unet_2d_condition.UNet2DConditionModel.attn_processors
  def 		lowerCamelCase__				(  self       :    str						)   ->    List[str]:


     '''simple docstring'''




     SCREAMING_SNAKE_CASE_:       List[str]   	={}

     def fn_recursive_add_processors(lowerCAmelCase       :    List[str]				,						lowerCAmelCase       :    str				,						lowerCAmelCase       :    Optional[int]						):
        if hasattr(__lowerCAmelCase				,						"""set_processor"""						):
           SCREAMING_SNAKE_CASE_:       Optional[int]   	=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 		lowerCamelCase__				(  self       :    Tuple				,						lowerCAmelCase       :    Optional[Any]						)   ->    Optional[Any]:


     '''simple docstring'''




     SCREAMING_SNAKE_CASE_:       Dict   	=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       :    List[Any]				,						lowerCAmelCase       :    str				,						lowerCAmelCase       :    Union[str, Any]						):
        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 		lowerCamelCase__				(  self       :    Optional[Any]						)   ->    Union[str, Any]:


     '''simple docstring'''




     self.set_attn_processor(AttnProcessor()						)



  @apply_forward_hook
  def 		lowerCamelCase__				(  self       :    str				,						lowerCAmelCase       :    List[str]				,						lowerCAmelCase       :    List[str] = True						)   ->    Tuple:


     '''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:
        SCREAMING_SNAKE_CASE_:       Dict   	=[self.encoder(__lowerCAmelCase						) for x_slice in x.split(1						)]
        SCREAMING_SNAKE_CASE_:       Any   	=torch.cat(__lowerCAmelCase						)
     else:
        SCREAMING_SNAKE_CASE_:       str   	=self.encoder(__lowerCAmelCase						)

     SCREAMING_SNAKE_CASE_:       int   	=self.quant_conv(__lowerCAmelCase						)
     SCREAMING_SNAKE_CASE_:       Union[str, Any]   	=DiagonalGaussianDistribution(__lowerCAmelCase						)

     if not return_dict:
        return (posterior,)

     return AutoencoderKLOutput(latent_dist=__lowerCAmelCase						)



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


     '''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						)

     SCREAMING_SNAKE_CASE_:       List[Any]   	=self.post_quant_conv(__lowerCAmelCase						)
     SCREAMING_SNAKE_CASE_:       List[Any]   	=self.decoder(__lowerCAmelCase						)

     if not return_dict:
        return (dec,)

     return DecoderOutput(sample=__lowerCAmelCase						)



  @apply_forward_hook
  def 		lowerCamelCase__				(  self       :    Optional[Any]				,						lowerCAmelCase       :    List[Any]				,						lowerCAmelCase       :    List[Any] = True						)   ->    List[Any]:


     '''simple docstring'''




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

     if not return_dict:
        return (decoded,)

     return DecoderOutput(sample=__lowerCAmelCase						)



  def 		lowerCamelCase__				(  self       :    Optional[int]				,						lowerCAmelCase       :    List[str]				,						lowerCAmelCase       :    List[Any]				,						lowerCAmelCase       :    Union[str, Any]						)   ->    Dict:


     '''simple docstring'''




     SCREAMING_SNAKE_CASE_:       Optional[int]   	=min(a.shape[2]				,						b.shape[2]				,						__lowerCAmelCase						)
     for y in range(__lowerCAmelCase						):
        SCREAMING_SNAKE_CASE_:       Tuple   	=a[:, :, -blend_extent + y, :] * (1 - y / blend_extent) + b[:, :, y, :] * (y / blend_extent)
     return b



  def 		lowerCamelCase__				(  self       :    int				,						lowerCAmelCase       :    Dict				,						lowerCAmelCase       :    Tuple				,						lowerCAmelCase       :    Any						)   ->    Tuple:


     '''simple docstring'''




     SCREAMING_SNAKE_CASE_:       List[str]   	=min(a.shape[3]				,						b.shape[3]				,						__lowerCAmelCase						)
     for x in range(__lowerCAmelCase						):
        SCREAMING_SNAKE_CASE_:       Optional[int]   	=a[:, :, :, -blend_extent + x] * (1 - x / blend_extent) + b[:, :, :, x] * (x / blend_extent)
     return b



  def 		lowerCamelCase__				(  self       :    Dict				,						lowerCAmelCase       :    Tuple				,						lowerCAmelCase       :    Any = True						)   ->    Optional[int]:


     '''simple docstring'''




     SCREAMING_SNAKE_CASE_:       Any   	=int(self.tile_sample_min_size * (1 - self.tile_overlap_factor)						)
     SCREAMING_SNAKE_CASE_:       Any   	=int(self.tile_latent_min_size * self.tile_overlap_factor						)
     SCREAMING_SNAKE_CASE_:       Any   	=self.tile_latent_min_size - blend_extent

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

     SCREAMING_SNAKE_CASE_:       Optional[int]   	=torch.cat(__lowerCAmelCase				,						dim=2						)
     SCREAMING_SNAKE_CASE_:       List[str]   	=DiagonalGaussianDistribution(__lowerCAmelCase						)

     if not return_dict:
        return (posterior,)

     return AutoencoderKLOutput(latent_dist=__lowerCAmelCase						)



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


     '''simple docstring'''




     SCREAMING_SNAKE_CASE_:       Any   	=int(self.tile_latent_min_size * (1 - self.tile_overlap_factor)						)
     SCREAMING_SNAKE_CASE_:       Any   	=int(self.tile_sample_min_size * self.tile_overlap_factor						)
     SCREAMING_SNAKE_CASE_:       Dict   	=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.
     SCREAMING_SNAKE_CASE_:       Dict   	=[]
     for i in range(0				,						z.shape[2]				,						__lowerCAmelCase						):
        SCREAMING_SNAKE_CASE_:       List[str]   	=[]
        for j in range(0				,						z.shape[3]				,						__lowerCAmelCase						):
           SCREAMING_SNAKE_CASE_:       List[Any]   	=z[:, :, i : i + self.tile_latent_min_size, j : j + self.tile_latent_min_size]
           SCREAMING_SNAKE_CASE_:       str   	=self.post_quant_conv(__lowerCAmelCase						)
           SCREAMING_SNAKE_CASE_:       Union[str, Any]   	=self.decoder(__lowerCAmelCase						)
           row.append(__lowerCAmelCase						)
        rows.append(__lowerCAmelCase						)
     SCREAMING_SNAKE_CASE_:       Optional[int]   	=[]
     for i, row in enumerate(__lowerCAmelCase						):
        SCREAMING_SNAKE_CASE_:       str   	=[]
        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:
              SCREAMING_SNAKE_CASE_:       Dict   	=self.blend_v(rows[i - 1][j]				,						__lowerCAmelCase				,						__lowerCAmelCase						)
           if j > 0:
              SCREAMING_SNAKE_CASE_:       int   	=self.blend_h(row[j - 1]				,						__lowerCAmelCase				,						__lowerCAmelCase						)
           result_row.append(tile[:, :, :row_limit, :row_limit]						)
        result_rows.append(torch.cat(__lowerCAmelCase				,						dim=3						)						)

     SCREAMING_SNAKE_CASE_:       Any   	=torch.cat(__lowerCAmelCase				,						dim=2						)
     if not return_dict:
        return (dec,)

     return DecoderOutput(sample=__lowerCAmelCase						)





  def 		lowerCamelCase__				(  self       :    List[Any]				,						lowerCAmelCase       :    Optional[int]				,						lowerCAmelCase       :    Any = False				,						lowerCAmelCase       :    Union[str, Any] = True				,						lowerCAmelCase       :    Union[str, Any] = None				,						)   ->    List[Any]:


     '''simple docstring'''




     SCREAMING_SNAKE_CASE_:       List[str]   	=sample
     SCREAMING_SNAKE_CASE_:       Tuple   	=self.encode(__lowerCAmelCase						).latent_dist
     if sample_posterior:
        SCREAMING_SNAKE_CASE_:       Optional[int]   	=posterior.sample(generator=__lowerCAmelCase						)
     else:
        SCREAMING_SNAKE_CASE_:       int   	=posterior.mode()
     SCREAMING_SNAKE_CASE_:       List[str]   	=self.decode(__lowerCAmelCase						).sample

     if not return_dict:
        return (dec,)

     return DecoderOutput(sample=__lowerCAmelCase						)




'''simple docstring'''




import unittest

import numpy as np

from transformers.file_utils import is_torch_available, is_vision_available
from transformers.testing_utils import require_torch, require_vision

from ...test_image_processing_common import ImageProcessingSavingTestMixin, prepare_image_inputs


if is_torch_available():
						import torch

if is_vision_available():
						from PIL import Image

						from transformers import DPTImageProcessor






class 				a__(					unittest.TestCase	):


			'''simple docstring'''






			def __init__(   self       ,    __lowerCAmelCase       ,    __lowerCAmelCase=7       ,    __lowerCAmelCase=3       ,    __lowerCAmelCase=18       ,    __lowerCAmelCase=30       ,    __lowerCAmelCase=400       ,    __lowerCAmelCase=True       ,    __lowerCAmelCase=None       ,    __lowerCAmelCase=True       ,    __lowerCAmelCase=[0.5, 0.5, 0.5]       ,    __lowerCAmelCase=[0.5, 0.5, 0.5]       ,    ):




					"""simple docstring"""


					lowerCAmelCase   					=  size if size is not None else {"""height""": 18, """width""": 18}
					lowerCAmelCase   					=  parent
					lowerCAmelCase   					=  batch_size
					lowerCAmelCase   					=  num_channels
					lowerCAmelCase   					=  image_size
					lowerCAmelCase   					=  min_resolution
					lowerCAmelCase   					=  max_resolution
					lowerCAmelCase   					=  do_resize
					lowerCAmelCase   					=  size
					lowerCAmelCase   					=  do_normalize
					lowerCAmelCase   					=  image_mean
					lowerCAmelCase   					=  image_std







			def 							a_       (   self):




					"""simple docstring"""


					return {
					    "image_mean": self.image_mean,
					    "image_std": self.image_std,
					    "do_normalize": self.do_normalize,
					    "do_resize": self.do_resize,
					    "size": self.size,
					}







@require_torch
@require_vision
class 				a__(					lowerCAmelCase__ ,						unittest.TestCase	):


			'''simple docstring'''






			UpperCAmelCase_			:     Tuple      =       DPTImageProcessor if is_vision_available() else None



			def 							a_       (   self):




					"""simple docstring"""


					lowerCAmelCase   					=  DPTImageProcessingTester(self)



			@property
			def 							a_       (   self):




					"""simple docstring"""


					return self.image_processor_tester.prepare_image_processor_dict()



			def 							a_       (   self):




					"""simple docstring"""


					lowerCAmelCase   					=  self.image_processing_class(**self.image_processor_dict)
					self.assertTrue(hasattr(__lowerCAmelCase       ,    """image_mean"""))
					self.assertTrue(hasattr(__lowerCAmelCase       ,    """image_std"""))
					self.assertTrue(hasattr(__lowerCAmelCase       ,    """do_normalize"""))
					self.assertTrue(hasattr(__lowerCAmelCase       ,    """do_resize"""))
					self.assertTrue(hasattr(__lowerCAmelCase       ,    """size"""))



			def 							a_       (   self):




					"""simple docstring"""


					lowerCAmelCase   					=  self.image_processing_class.from_dict(self.image_processor_dict)
					self.assertEqual(image_processor.size       ,    {"""height""": 18, """width""": 18})

					lowerCAmelCase   					=  self.image_processing_class.from_dict(self.image_processor_dict       ,    size=42)
					self.assertEqual(image_processor.size       ,    {"""height""": 42, """width""": 42})



			def 							a_       (   self):




					"""simple docstring"""


					lowerCAmelCase   					=  self.image_processing_class(**self.image_processor_dict)
					# create random PIL images
					lowerCAmelCase   					=  prepare_image_inputs(self.image_processor_tester       ,    equal_resolution=__lowerCAmelCase)
					for image in image_inputs:
							self.assertIsInstance(__lowerCAmelCase       ,    Image.Image)

					# Test not batched input
					lowerCAmelCase   					=  image_processing(image_inputs[0]       ,    return_tensors="""pt""").pixel_values
					self.assertEqual(
					    encoded_images.shape       ,    (
					        1,
					        self.image_processor_tester.num_channels,
					        self.image_processor_tester.size["""height"""],
					        self.image_processor_tester.size["""width"""],
					    )       ,    )

					# Test batched
					lowerCAmelCase   					=  image_processing(__lowerCAmelCase       ,    return_tensors="""pt""").pixel_values
					self.assertEqual(
					    encoded_images.shape       ,    (
					        self.image_processor_tester.batch_size,
					        self.image_processor_tester.num_channels,
					        self.image_processor_tester.size["""height"""],
					        self.image_processor_tester.size["""width"""],
					    )       ,    )



			def 							a_       (   self):




					"""simple docstring"""


					lowerCAmelCase   					=  self.image_processing_class(**self.image_processor_dict)
					# create random numpy tensors
					lowerCAmelCase   					=  prepare_image_inputs(self.image_processor_tester       ,    equal_resolution=__lowerCAmelCase       ,    numpify=__lowerCAmelCase)
					for image in image_inputs:
							self.assertIsInstance(__lowerCAmelCase       ,    np.ndarray)

					# Test not batched input
					lowerCAmelCase   					=  image_processing(image_inputs[0]       ,    return_tensors="""pt""").pixel_values
					self.assertEqual(
					    encoded_images.shape       ,    (
					        1,
					        self.image_processor_tester.num_channels,
					        self.image_processor_tester.size["""height"""],
					        self.image_processor_tester.size["""width"""],
					    )       ,    )

					# Test batched
					lowerCAmelCase   					=  image_processing(__lowerCAmelCase       ,    return_tensors="""pt""").pixel_values
					self.assertEqual(
					    encoded_images.shape       ,    (
					        self.image_processor_tester.batch_size,
					        self.image_processor_tester.num_channels,
					        self.image_processor_tester.size["""height"""],
					        self.image_processor_tester.size["""width"""],
					    )       ,    )







			def 							a_       (   self):




					"""simple docstring"""


					lowerCAmelCase   					=  self.image_processing_class(**self.image_processor_dict)
					# create random PyTorch tensors
					lowerCAmelCase   					=  prepare_image_inputs(self.image_processor_tester       ,    equal_resolution=__lowerCAmelCase       ,    torchify=__lowerCAmelCase)
					for image in image_inputs:
							self.assertIsInstance(__lowerCAmelCase       ,    torch.Tensor)

					# Test not batched input
					lowerCAmelCase   					=  image_processing(image_inputs[0]       ,    return_tensors="""pt""").pixel_values
					self.assertEqual(
					    encoded_images.shape       ,    (
					        1,
					        self.image_processor_tester.num_channels,
					        self.image_processor_tester.size["""height"""],
					        self.image_processor_tester.size["""width"""],
					    )       ,    )

					# Test batched
					lowerCAmelCase   					=  image_processing(__lowerCAmelCase       ,    return_tensors="""pt""").pixel_values
					self.assertEqual(
					    encoded_images.shape       ,    (
					        self.image_processor_tester.batch_size,
					        self.image_processor_tester.num_channels,
					        self.image_processor_tester.size["""height"""],
					        self.image_processor_tester.size["""width"""],
					    )       ,    )




