





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       (	__lowercase						):
      _snake_case							     =42

class      A       (	__lowercase		,      __lowercase						):
      _snake_case							     =True



      @register_to_config
      def __init__(	self:						Dict		,  _lowerCAmelCase:						int = 3		,  _lowerCAmelCase:						int = 3		,  _lowerCAmelCase:						Tuple[str] = ("DownEncoderBlock2D",)		,  _lowerCAmelCase:						Tuple[str] = ("UpDecoderBlock2D",)		,  _lowerCAmelCase:						Tuple[int] = (64,)		,  _lowerCAmelCase:						int = 1		,  _lowerCAmelCase:						str = "silu"		,  _lowerCAmelCase:						int = 4		,  _lowerCAmelCase:						int = 32		,  _lowerCAmelCase:						int = 32		,  _lowerCAmelCase:						float = 0.1_82_15		,  )						->    List[str]:
         '''simple docstring'''

         super().__init__()

         # pass init params to Encoder
         UpperCAmelCase_											=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
         UpperCAmelCase_											=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		,  )

         UpperCAmelCase_											=nn.Convad(2 * latent_channels		,  2 * latent_channels		,  1       )
         UpperCAmelCase_											=nn.Convad(_lowerCAmelCase		,  _lowerCAmelCase		,  1       )

         UpperCAmelCase_											=False
         UpperCAmelCase_											=False

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



      def 	lowerCAmelCase__    (	self:						str		,  _lowerCAmelCase:						Union[str, Any]		,  _lowerCAmelCase:						Dict=False       )						->    Any:
         '''simple docstring'''

         if isinstance(_lowerCAmelCase		,  (Encoder, Decoder)       ):
            UpperCAmelCase_											=value



      def 	lowerCAmelCase__    (	self:						Optional[Any]		,  _lowerCAmelCase:						bool = True       )						->    List[str]:
         '''simple docstring'''

         UpperCAmelCase_											=use_tiling



      def 	lowerCAmelCase__    (	self:						Optional[int]       )						->    Any:
         '''simple docstring'''

         self.enable_tiling(_lowerCAmelCase       )



      def 	lowerCAmelCase__    (	self:						Any       )						->    str:
         '''simple docstring'''

         UpperCAmelCase_											=True



      def 	lowerCAmelCase__    (	self:						int       )						->    Optional[int]:
         '''simple docstring'''

         UpperCAmelCase_											=False



      @property
      # Copied from diffusers.models.unet_2d_condition.UNet2DConditionModel.attn_processors
      def 	lowerCAmelCase__    (	self:						Optional[int]       )						->    Dict[str, AttentionProcessor]:
         '''simple docstring'''

         UpperCAmelCase_											={}

         def fn_recursive_add_processors(_lowerCAmelCase:						str		,  _lowerCAmelCase:						torch.nn.Module		,  _lowerCAmelCase:						Dict[str, AttentionProcessor]       ):
            if hasattr(_lowerCAmelCase		,  "set_processor"       ):
               UpperCAmelCase_											=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:						List[str]		,  _lowerCAmelCase:						Union[AttentionProcessor, Dict[str, AttentionProcessor]]       )						->    Optional[Any]:
         '''simple docstring'''

         UpperCAmelCase_											=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:						str		,  _lowerCAmelCase:						torch.nn.Module		,  _lowerCAmelCase:						Tuple       ):
            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:						Dict       )						->    List[str]:
         '''simple docstring'''

         self.set_attn_processor(AttnProcessor()       )



      @apply_forward_hook
      def 	lowerCAmelCase__    (	self:						Dict		,  _lowerCAmelCase:						torch.FloatTensor		,  _lowerCAmelCase:						bool = 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:
            UpperCAmelCase_											=[self.encoder(_lowerCAmelCase       ) for x_slice in x.split(1       )]
            UpperCAmelCase_											=torch.cat(_lowerCAmelCase       )
         else:
            UpperCAmelCase_											=self.encoder(_lowerCAmelCase       )

         UpperCAmelCase_											=self.quant_conv(_lowerCAmelCase       )
         UpperCAmelCase_											=DiagonalGaussianDistribution(_lowerCAmelCase       )

         if not return_dict:
            return (posterior,)

         return AutoencoderKLOutput(latent_dist=_lowerCAmelCase       )



      def 	lowerCAmelCase__    (	self:						Tuple		,  _lowerCAmelCase:						torch.FloatTensor		,  _lowerCAmelCase:						bool = 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       )

         UpperCAmelCase_											=self.post_quant_conv(_lowerCAmelCase       )
         UpperCAmelCase_											=self.decoder(_lowerCAmelCase       )

         if not return_dict:
            return (dec,)

         return DecoderOutput(sample=_lowerCAmelCase       )



      @apply_forward_hook
      def 	lowerCAmelCase__    (	self:						int		,  _lowerCAmelCase:						torch.FloatTensor		,  _lowerCAmelCase:						bool = True       )						->    Union[DecoderOutput, torch.FloatTensor]:
         '''simple docstring'''

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

         if not return_dict:
            return (decoded,)

         return DecoderOutput(sample=_lowerCAmelCase       )



      def 	lowerCAmelCase__    (	self:						List[Any]		,  _lowerCAmelCase:						Union[str, Any]		,  _lowerCAmelCase:						str		,  _lowerCAmelCase:						Optional[Any]       )						->    List[str]:
         '''simple docstring'''

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



      def 	lowerCAmelCase__    (	self:						Any		,  _lowerCAmelCase:						str		,  _lowerCAmelCase:						List[str]		,  _lowerCAmelCase:						Optional[Any]       )						->    str:
         '''simple docstring'''

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



      def 	lowerCAmelCase__    (	self:						Union[str, Any]		,  _lowerCAmelCase:						torch.FloatTensor		,  _lowerCAmelCase:						bool = True       )						->    AutoencoderKLOutput:
         '''simple docstring'''

         UpperCAmelCase_											=int(self.tile_sample_min_size * (1 - self.tile_overlap_factor)       )
         UpperCAmelCase_											=int(self.tile_latent_min_size * self.tile_overlap_factor       )
         UpperCAmelCase_											=self.tile_latent_min_size - blend_extent

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

         UpperCAmelCase_											=torch.cat(_lowerCAmelCase		,  dim=2       )
         UpperCAmelCase_											=DiagonalGaussianDistribution(_lowerCAmelCase       )

         if not return_dict:
            return (posterior,)

         return AutoencoderKLOutput(latent_dist=_lowerCAmelCase       )



      def 	lowerCAmelCase__    (	self:						str		,  _lowerCAmelCase:						torch.FloatTensor		,  _lowerCAmelCase:						bool = True       )						->    Union[DecoderOutput, torch.FloatTensor]:
         '''simple docstring'''

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

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

         return DecoderOutput(sample=_lowerCAmelCase       )






      def 	lowerCAmelCase__    (	self:						List[Any]		,  _lowerCAmelCase:						torch.FloatTensor		,  _lowerCAmelCase:						bool = False		,  _lowerCAmelCase:						bool = True		,  _lowerCAmelCase:						Optional[torch.Generator] = None		,  )						->    Union[DecoderOutput, torch.FloatTensor]:
         '''simple docstring'''

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

         if not return_dict:
            return (dec,)

         return DecoderOutput(sample=_lowerCAmelCase       )











import sys

from .dependency_versions_table import deps
from .utils.versions import require_version, require_version_core


# define which module versions we always want to check at run time
# (usually the ones defined in `install_requires` in setup.py)
#
# order specific notes:
# - tqdm must be checked before tokenizers

__lowercase  :   Optional[int]						      ="""python tqdm regex requests packaging filelock numpy tokenizers""".split()
if sys.version_info < (3, 7):
     pkgs_to_check_at_runtime.append("""dataclasses""")
if sys.version_info < (3, 8):
     pkgs_to_check_at_runtime.append("""importlib_metadata""")

for pkg in pkgs_to_check_at_runtime:
     if pkg in deps:
          if pkg == "tokenizers":
               # must be loaded here, or else tqdm check may fail
               from .utils import is_tokenizers_available

               if not is_tokenizers_available():
                    continue  # not required, check version only if installed

          require_version_core(deps[pkg])
     else:
          raise ValueError(f"""can't find {pkg} in {deps.keys()}, check dependency_versions_table.py""")




def     a__ (					lowercase__  ,    lowercase__=None  ):






   '''simple docstring'''






   require_version(deps[pkg]  ,    lowercase__  )




