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

     """simple docstring"""

     a_    	=					42







class  _UpperCAmelCase ( lowerCAmelCase__	,lowerCAmelCase__    ):

     """simple docstring"""

     a_    	=					True


     @register_to_config
     def __init__(      self							,		lowerCAmelCase_ = 3							,		lowerCAmelCase_ = 3							,		lowerCAmelCase_ = ("DownEncoderBlock2D",)							,		lowerCAmelCase_ = ("UpDecoderBlock2D",)							,		lowerCAmelCase_ = (64,)							,		lowerCAmelCase_ = 1							,		lowerCAmelCase_ = "silu"							,		lowerCAmelCase_ = 4							,		lowerCAmelCase_ = 32							,		lowerCAmelCase_ = 32							,		lowerCAmelCase_ = 0.18215							,		):

       '''simple docstring'''

       super().__init__()

       # pass init params to Encoder
       a_      :	Optional[Any]      =							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_      :	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_							,		)

       a_      :	Dict      =							nn.Convad(2 * latent_channels							,		2 * latent_channels							,		1 )
       a_      :	Tuple      =							nn.Convad(lowerCAmelCase_							,		lowerCAmelCase_							,		1 )

       a_      :	Any      =							False
       a_      :	Optional[int]      =							False

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


     def 							_lowerCAmelCase  (      self							,		lowerCAmelCase_							,		lowerCAmelCase_=False ):

       '''simple docstring'''

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


     def 							_lowerCAmelCase  (      self							,		lowerCAmelCase_ = True ):

       '''simple docstring'''

       a_      :	Optional[Any]      =							use_tiling


     def 							_lowerCAmelCase  (      self ):

       '''simple docstring'''

       self.enable_tiling(lowerCAmelCase_ )


     def 							_lowerCAmelCase  (      self ):

       '''simple docstring'''

       a_      :	Optional[int]      =							True


     def 							_lowerCAmelCase  (      self ):

       '''simple docstring'''

       a_      :	List[str]      =							False


     @property
     # Copied from diffusers.models.unet_2d_condition.UNet2DConditionModel.attn_processors
     def 							_lowerCAmelCase  (      self ):

       '''simple docstring'''

       a_      :	Union[str, Any]      =							{}

       def fn_recursive_add_processors(lowerCAmelCase_							,		lowerCAmelCase_							,		lowerCAmelCase_ ):
         if hasattr(lowerCAmelCase_							,		"""set_processor""" ):
           a_      :	Union[str, Any]      =							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							,		lowerCAmelCase_ ):

       '''simple docstring'''

       a_      :	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_							,		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 							_lowerCAmelCase  (      self ):

       '''simple docstring'''

       self.set_attn_processor(AttnProcessor() )


     @apply_forward_hook
     def 							_lowerCAmelCase  (      self							,		lowerCAmelCase_							,		lowerCAmelCase_ = True ):

       '''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_      :	Tuple      =							[self.encoder(lowerCAmelCase_ ) for x_slice in x.split(1 )]
         a_      :	str      =							torch.cat(lowerCAmelCase_ )
       else:
         a_      :	int      =							self.encoder(lowerCAmelCase_ )

       a_      :	int      =							self.quant_conv(lowerCAmelCase_ )
       a_      :	List[str]      =							DiagonalGaussianDistribution(lowerCAmelCase_ )

       if not return_dict:
         return (posterior,)

       return AutoencoderKLOutput(latent_dist=lowerCAmelCase_ )


     def 							_lowerCAmelCase  (      self							,		lowerCAmelCase_							,		lowerCAmelCase_ = True ):

       '''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_      :	str      =							self.post_quant_conv(lowerCAmelCase_ )
       a_      :	Any      =							self.decoder(lowerCAmelCase_ )

       if not return_dict:
         return (dec,)

       return DecoderOutput(sample=lowerCAmelCase_ )


     @apply_forward_hook
     def 							_lowerCAmelCase  (      self							,		lowerCAmelCase_							,		lowerCAmelCase_ = True ):

       '''simple docstring'''

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

       if not return_dict:
         return (decoded,)

       return DecoderOutput(sample=lowerCAmelCase_ )


     def 							_lowerCAmelCase  (      self							,		lowerCAmelCase_							,		lowerCAmelCase_							,		lowerCAmelCase_ ):

       '''simple docstring'''

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


     def 							_lowerCAmelCase  (      self							,		lowerCAmelCase_							,		lowerCAmelCase_							,		lowerCAmelCase_ ):

       '''simple docstring'''

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


     def 							_lowerCAmelCase  (      self							,		lowerCAmelCase_							,		lowerCAmelCase_ = True ):

       '''simple docstring'''

       a_      :	int      =							int(self.tile_sample_min_size * (1 - self.tile_overlap_factor) )
       a_      :	Union[str, Any]      =							int(self.tile_latent_min_size * self.tile_overlap_factor )
       a_      :	Any      =							self.tile_latent_min_size - blend_extent

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

       a_      :	Dict      =							torch.cat(lowerCAmelCase_							,		dim=2 )
       a_      :	Optional[int]      =							DiagonalGaussianDistribution(lowerCAmelCase_ )

       if not return_dict:
         return (posterior,)

       return AutoencoderKLOutput(latent_dist=lowerCAmelCase_ )


     def 							_lowerCAmelCase  (      self							,		lowerCAmelCase_							,		lowerCAmelCase_ = True ):

       '''simple docstring'''

       a_      :	str      =							int(self.tile_latent_min_size * (1 - self.tile_overlap_factor) )
       a_      :	Dict      =							int(self.tile_sample_min_size * self.tile_overlap_factor )
       a_      :	str      =							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_      :	Dict      =							[]
       for i in range(0							,		z.shape[2]							,		lowerCAmelCase_ ):
         a_      :	List[Any]      =							[]
         for j in range(0							,		z.shape[3]							,		lowerCAmelCase_ ):
           a_      :	Any      =							z[:, :, i : i + self.tile_latent_min_size, j : j + self.tile_latent_min_size]
           a_      :	Any      =							self.post_quant_conv(lowerCAmelCase_ )
           a_      :	List[str]      =							self.decoder(lowerCAmelCase_ )
           row.append(lowerCAmelCase_ )
         rows.append(lowerCAmelCase_ )
       a_      :	List[str]      =							[]
       for i, row in enumerate(lowerCAmelCase_ ):
         a_      :	int      =							[]
         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_      :	List[str]      =							self.blend_v(rows[i - 1][j]							,		lowerCAmelCase_							,		lowerCAmelCase_ )
           if j > 0:
             a_      :	Union[str, 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 ) )

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

       return DecoderOutput(sample=lowerCAmelCase_ )






     def 							_lowerCAmelCase  (      self							,		lowerCAmelCase_							,		lowerCAmelCase_ = False							,		lowerCAmelCase_ = True							,		lowerCAmelCase_ = None							,		):

       '''simple docstring'''

       a_      :	int      =							sample
       a_      :	int      =							self.encode(lowerCAmelCase_ ).latent_dist
       if sample_posterior:
         a_      :	Optional[Any]      =							posterior.sample(generator=lowerCAmelCase_ )
       else:
         a_      :	Tuple      =							posterior.mode()
       a_      :	Optional[int]      =							self.decode(lowerCAmelCase_ ).sample

       if not return_dict:
         return (dec,)

       return DecoderOutput(sample=lowerCAmelCase_ )




'''simple docstring'''

from typing import TYPE_CHECKING

from ...utils import (
    OptionalDependencyNotAvailable,
    _LazyModule,
    is_flax_available,
    is_tf_available,
    is_torch_available,
)


__snake_case:  Dict								=			{"configuration_encoder_decoder": ["EncoderDecoderConfig"]}

try:
    if not is_torch_available():
        raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
    pass
else:
    __snake_case:  List[str]								=			["EncoderDecoderModel"]

try:
    if not is_tf_available():
        raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
    pass
else:
    __snake_case:  Any								=			["TFEncoderDecoderModel"]

try:
    if not is_flax_available():
        raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
    pass
else:
    __snake_case:  List[str]								=			["FlaxEncoderDecoderModel"]

if TYPE_CHECKING:
    from .configuration_encoder_decoder import EncoderDecoderConfig

    try:
        if not is_torch_available():
            raise OptionalDependencyNotAvailable()
    except OptionalDependencyNotAvailable:
        pass
    else:
        from .modeling_encoder_decoder import EncoderDecoderModel

    try:
        if not is_tf_available():
            raise OptionalDependencyNotAvailable()
    except OptionalDependencyNotAvailable:
        pass
    else:
        from .modeling_tf_encoder_decoder import TFEncoderDecoderModel

    try:
        if not is_flax_available():
            raise OptionalDependencyNotAvailable()
    except OptionalDependencyNotAvailable:
        pass
    else:
        from .modeling_flax_encoder_decoder import FlaxEncoderDecoderModel

else:
    import sys

    __snake_case:  Tuple								=			_LazyModule(__name__, globals()["__file__"], _import_structure, module_spec=__spec__)


