


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  lowerCAmelCase_						(       snake_case__      ):





  """simple docstring"""



  a_       :"DiagonalGaussianDistribution"


class  lowerCAmelCase_						(       snake_case__		,    snake_case__      ):





  """simple docstring"""



  a_       :Optional[int]												=True

  @register_to_config
  def __init__(					self     :							Tuple				,							SCREAMING_SNAKE_CASE__     :							int = 3				,							SCREAMING_SNAKE_CASE__     :							int = 3				,							SCREAMING_SNAKE_CASE__     :							Tuple[str] = ("DownEncoderBlock2D",)				,							SCREAMING_SNAKE_CASE__     :							Tuple[str] = ("UpDecoderBlock2D",)				,							SCREAMING_SNAKE_CASE__     :							Tuple[int] = (6_4,)				,							SCREAMING_SNAKE_CASE__     :							int = 1				,							SCREAMING_SNAKE_CASE__     :							str = "silu"				,							SCREAMING_SNAKE_CASE__     :							int = 4				,							SCREAMING_SNAKE_CASE__     :							int = 3_2				,							SCREAMING_SNAKE_CASE__     :							int = 3_2				,							SCREAMING_SNAKE_CASE__     :							float = 0.1_8_2_1_5				,							):





       '''simple docstring'''

       super().__init__()

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

       __a       =   nn.Convad(2 * latent_channels				,							2 * latent_channels				,							1						)
       __a       =   nn.Convad(SCREAMING_SNAKE_CASE__				,							SCREAMING_SNAKE_CASE__				,							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 __a						(					self     :							Optional[int]				,							SCREAMING_SNAKE_CASE__     :							Dict				,							SCREAMING_SNAKE_CASE__     :							Dict=False						):





       '''simple docstring'''

       if isinstance(SCREAMING_SNAKE_CASE__				,							(Encoder, Decoder)						):
            __a       =   value

  def __a						(					self     :							Optional[int]				,							SCREAMING_SNAKE_CASE__     :							bool = True						):





       '''simple docstring'''

       __a       =   use_tiling

  def __a						(					self     :							int						):





       '''simple docstring'''

       self.enable_tiling(SCREAMING_SNAKE_CASE__						)

  def __a						(					self     :							Optional[int]						):





       '''simple docstring'''

       __a       =   True

  def __a						(					self     :							List[Any]						):





       '''simple docstring'''

       __a       =   False

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





       '''simple docstring'''

       __a       =   {}

       def fn_recursive_add_processors(SCREAMING_SNAKE_CASE__     :							str				,							SCREAMING_SNAKE_CASE__     :							torch.nn.Module				,							SCREAMING_SNAKE_CASE__     :							Dict[str, AttentionProcessor]						):
            if hasattr(SCREAMING_SNAKE_CASE__				,							"""set_processor"""						):
                 __a       =   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     :							Optional[Any]				,							SCREAMING_SNAKE_CASE__     :							Union[AttentionProcessor, Dict[str, AttentionProcessor]]						):





       '''simple docstring'''

       __a       =   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__     :							str				,							SCREAMING_SNAKE_CASE__     :							torch.nn.Module				,							SCREAMING_SNAKE_CASE__     :							Optional[Any]						):
            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     :							int						):





       '''simple docstring'''

       self.set_attn_processor(AttnProcessor()						)

  @apply_forward_hook
  def __a						(					self     :							List[str]				,							SCREAMING_SNAKE_CASE__     :							torch.FloatTensor				,							SCREAMING_SNAKE_CASE__     :							bool = 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(SCREAMING_SNAKE_CASE__				,							return_dict=SCREAMING_SNAKE_CASE__						)

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

       __a       =   self.quant_conv(SCREAMING_SNAKE_CASE__						)
       __a       =   DiagonalGaussianDistribution(SCREAMING_SNAKE_CASE__						)

       if not return_dict:
            return (posterior,)

       return AutoencoderKLOutput(latent_dist=SCREAMING_SNAKE_CASE__						)

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

       __a       =   self.post_quant_conv(SCREAMING_SNAKE_CASE__						)
       __a       =   self.decoder(SCREAMING_SNAKE_CASE__						)

       if not return_dict:
            return (dec,)

       return DecoderOutput(sample=SCREAMING_SNAKE_CASE__						)

  @apply_forward_hook
  def __a						(					self     :							str				,							SCREAMING_SNAKE_CASE__     :							torch.FloatTensor				,							SCREAMING_SNAKE_CASE__     :							bool = True						):





       '''simple docstring'''

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

       if not return_dict:
            return (decoded,)

       return DecoderOutput(sample=SCREAMING_SNAKE_CASE__						)

  def __a						(					self     :							str				,							SCREAMING_SNAKE_CASE__     :							Dict				,							SCREAMING_SNAKE_CASE__     :							int				,							SCREAMING_SNAKE_CASE__     :							int						):





       '''simple docstring'''

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

  def __a						(					self     :							List[Any]				,							SCREAMING_SNAKE_CASE__     :							int				,							SCREAMING_SNAKE_CASE__     :							Any				,							SCREAMING_SNAKE_CASE__     :							Tuple						):





       '''simple docstring'''

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

  def __a						(					self     :							List[str]				,							SCREAMING_SNAKE_CASE__     :							torch.FloatTensor				,							SCREAMING_SNAKE_CASE__     :							bool = True						):





       '''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]				,							SCREAMING_SNAKE_CASE__						):
            __a       =   []
            for j in range(0				,							x.shape[3]				,							SCREAMING_SNAKE_CASE__						):
                 __a       =   x[:, :, i : i + self.tile_sample_min_size, j : j + self.tile_sample_min_size]
                 __a       =   self.encoder(SCREAMING_SNAKE_CASE__						)
                 __a       =   self.quant_conv(SCREAMING_SNAKE_CASE__						)
                 row.append(SCREAMING_SNAKE_CASE__						)
            rows.append(SCREAMING_SNAKE_CASE__						)
       __a       =   []
       for i, row in enumerate(SCREAMING_SNAKE_CASE__						):
            __a       =   []
            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:
                      __a       =   self.blend_v(rows[i - 1][j]				,							SCREAMING_SNAKE_CASE__				,							SCREAMING_SNAKE_CASE__						)
                 if j > 0:
                      __a       =   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						)						)

       __a       =   torch.cat(SCREAMING_SNAKE_CASE__				,							dim=2						)
       __a       =   DiagonalGaussianDistribution(SCREAMING_SNAKE_CASE__						)

       if not return_dict:
            return (posterior,)

       return AutoencoderKLOutput(latent_dist=SCREAMING_SNAKE_CASE__						)

  def __a						(					self     :							Union[str, Any]				,							SCREAMING_SNAKE_CASE__     :							torch.FloatTensor				,							SCREAMING_SNAKE_CASE__     :							bool = True						):





       '''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]				,							SCREAMING_SNAKE_CASE__						):
            __a       =   []
            for j in range(0				,							z.shape[3]				,							SCREAMING_SNAKE_CASE__						):
                 __a       =   z[:, :, i : i + self.tile_latent_min_size, j : j + self.tile_latent_min_size]
                 __a       =   self.post_quant_conv(SCREAMING_SNAKE_CASE__						)
                 __a       =   self.decoder(SCREAMING_SNAKE_CASE__						)
                 row.append(SCREAMING_SNAKE_CASE__						)
            rows.append(SCREAMING_SNAKE_CASE__						)
       __a       =   []
       for i, row in enumerate(SCREAMING_SNAKE_CASE__						):
            __a       =   []
            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:
                      __a       =   self.blend_v(rows[i - 1][j]				,							SCREAMING_SNAKE_CASE__				,							SCREAMING_SNAKE_CASE__						)
                 if j > 0:
                      __a       =   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						)						)

       __a       =   torch.cat(SCREAMING_SNAKE_CASE__				,							dim=2						)
       if not return_dict:
            return (dec,)

       return DecoderOutput(sample=SCREAMING_SNAKE_CASE__						)


  def __a						(					self     :							Any				,							SCREAMING_SNAKE_CASE__     :							torch.FloatTensor				,							SCREAMING_SNAKE_CASE__     :							bool = False				,							SCREAMING_SNAKE_CASE__     :							bool = True				,							SCREAMING_SNAKE_CASE__     :							Optional[torch.Generator] = None				,							):





       '''simple docstring'''

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

       if not return_dict:
            return (dec,)

       return DecoderOutput(sample=SCREAMING_SNAKE_CASE__						)










'''simple docstring'''







import enum
import shutil
import sys


SCREAMING_SNAKE_CASE_      ,     SCREAMING_SNAKE_CASE_      =			shutil.get_terminal_size()

SCREAMING_SNAKE_CASE_      =			{'UP': 'A', 'DOWN': 'B', 'RIGHT': 'C', 'LEFT': 'D'}


class  lowerCAmelCase_						(       enum.Enum      ):





  """simple docstring"""



  a_       :List[Any]												=0
  a_       :Optional[Any]												=1




def        __lowercase						(      __SCREAMING_SNAKE_CASE				,  __SCREAMING_SNAKE_CASE=""		)				->					List[Any]:
     """simple docstring"""




     sys.stdout.write(str(__SCREAMING_SNAKE_CASE		) + end		)
     sys.stdout.flush()




def        __lowercase						(      __SCREAMING_SNAKE_CASE				,  __SCREAMING_SNAKE_CASE				,  __SCREAMING_SNAKE_CASE=""		)				->					Tuple:
     """simple docstring"""




     forceWrite(F'''\u001b[{color}m{content}\u001b[0m'''				,  __SCREAMING_SNAKE_CASE		)




def        __lowercase						(      )				->					Union[str, Any]:
     """simple docstring"""




     forceWrite("""\r"""		)




def        __lowercase						(      __SCREAMING_SNAKE_CASE				,  __SCREAMING_SNAKE_CASE		)				->					str:
     """simple docstring"""




     forceWrite(F'''\033[{num_lines}{CURSOR_TO_CHAR[direction.upper()]}'''		)




def        __lowercase						(      )				->					int:
     """simple docstring"""




     forceWrite(""" """ * TERMINAL_WIDTH		)
     reset_cursor()




def        __lowercase						(      )				->					str:
     """simple docstring"""




     reset_cursor()
     forceWrite("""-""" * TERMINAL_WIDTH		)






