

'''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__ ( snake_case_     ):
    lowercase		   =						42


class    A__ ( snake_case_  ,  snake_case_     ):
    lowercase		   =						True



    @register_to_config
    def __init__(   self						,			UpperCamelCase__ = 3						,			UpperCamelCase__ = 3						,			UpperCamelCase__ = ("DownEncoderBlock2D",)						,			UpperCamelCase__ = ("UpDecoderBlock2D",)						,			UpperCamelCase__ = (64,)						,			UpperCamelCase__ = 1						,			UpperCamelCase__ = "silu"						,			UpperCamelCase__ = 4						,			UpperCamelCase__ = 32						,			UpperCamelCase__ = 32						,			UpperCamelCase__ = 0.18215						,			)  -> Optional[Any]:





           '''simple docstring'''
           super().__init__()

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

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

           A_		   = nn.Convad(2 * latent_channels						,			2 * latent_channels						,			1    )
           A_		   = nn.Convad(UpperCamelCase__						,			UpperCamelCase__						,			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.25



    def    snake_case_     (   self						,			UpperCamelCase__						,			UpperCamelCase__=False    )  -> Tuple:





           '''simple docstring'''
           if isinstance(UpperCamelCase__						,			(Encoder, Decoder)    ):
                  A_		   = value



    def    snake_case_     (   self						,			UpperCamelCase__ = True    )  -> List[str]:





           '''simple docstring'''
           A_		   = use_tiling



    def    snake_case_     (   self    )  -> Dict:





           '''simple docstring'''
           self.enable_tiling(UpperCamelCase__    )



    def    snake_case_     (   self    )  -> Tuple:





           '''simple docstring'''
           A_		   = True



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





           '''simple docstring'''
           A_		   = False



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





           '''simple docstring'''
           A_		   = {}

           def fn_recursive_add_processors(UpperCamelCase__						,			UpperCamelCase__						,			UpperCamelCase__    ):
                  if hasattr(UpperCamelCase__						,			"""set_processor"""    ):
                         A_		   = module.processor

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

                  return processors

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

           return processors



    def    snake_case_     (   self						,			UpperCamelCase__    )  -> Optional[int]:





           '''simple docstring'''
           A_		   = len(self.attn_processors.keys()    )

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

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

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



    def    snake_case_     (   self    )  -> Tuple:





           '''simple docstring'''
           self.set_attn_processor(AttnProcessor()    )



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

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

           A_		   = self.quant_conv(UpperCamelCase__    )
           A_		   = DiagonalGaussianDistribution(UpperCamelCase__    )

           if not return_dict:
                  return (posterior,)

           return AutoencoderKLOutput(latent_dist=UpperCamelCase__    )



    def    snake_case_     (   self						,			UpperCamelCase__						,			UpperCamelCase__ = 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(UpperCamelCase__						,			return_dict=UpperCamelCase__    )

           A_		   = self.post_quant_conv(UpperCamelCase__    )
           A_		   = self.decoder(UpperCamelCase__    )

           if not return_dict:
                  return (dec,)

           return DecoderOutput(sample=UpperCamelCase__    )



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





           '''simple docstring'''
           if self.use_slicing and z.shape[0] > 1:
                  A_		   = [self._decode(UpperCamelCase__    ).sample for z_slice in z.split(1    )]
                  A_		   = torch.cat(UpperCamelCase__    )
           else:
                  A_		   = self._decode(UpperCamelCase__    ).sample

           if not return_dict:
                  return (decoded,)

           return DecoderOutput(sample=UpperCamelCase__    )



    def    snake_case_     (   self						,			UpperCamelCase__						,			UpperCamelCase__						,			UpperCamelCase__    )  -> Any:





           '''simple docstring'''
           A_		   = min(a.shape[2]						,			b.shape[2]						,			UpperCamelCase__    )
           for y in range(UpperCamelCase__    ):
                  A_		   = a[:, :, -blend_extent + y, :] * (1 - y / blend_extent) + b[:, :, y, :] * (y / blend_extent)
           return b



    def    snake_case_     (   self						,			UpperCamelCase__						,			UpperCamelCase__						,			UpperCamelCase__    )  -> str:





           '''simple docstring'''
           A_		   = min(a.shape[3]						,			b.shape[3]						,			UpperCamelCase__    )
           for x in range(UpperCamelCase__    ):
                  A_		   = a[:, :, :, -blend_extent + x] * (1 - x / blend_extent) + b[:, :, :, x] * (x / blend_extent)
           return b



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

           A_		   = torch.cat(UpperCamelCase__						,			dim=2    )
           A_		   = DiagonalGaussianDistribution(UpperCamelCase__    )

           if not return_dict:
                  return (posterior,)

           return AutoencoderKLOutput(latent_dist=UpperCamelCase__    )



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

           A_		   = torch.cat(UpperCamelCase__						,			dim=2    )
           if not return_dict:
                  return (dec,)

           return DecoderOutput(sample=UpperCamelCase__    )







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





           '''simple docstring'''
           A_		   = sample
           A_		   = self.encode(UpperCamelCase__    ).latent_dist
           if sample_posterior:
                  A_		   = posterior.sample(generator=UpperCamelCase__    )
           else:
                  A_		   = posterior.mode()
           A_		   = self.decode(UpperCamelCase__    ).sample

           if not return_dict:
                  return (dec,)

           return DecoderOutput(sample=UpperCamelCase__    )



'''simple docstring'''







from typing import TYPE_CHECKING

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


__lowerCamelCase						    =      {
    '''configuration_blenderbot''': [
        '''BLENDERBOT_PRETRAINED_CONFIG_ARCHIVE_MAP''',
        '''BlenderbotConfig''',
        '''BlenderbotOnnxConfig''',
    ],
    '''tokenization_blenderbot''': ['''BlenderbotTokenizer'''],
}

try:
      if not is_tokenizers_available():
            raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
      pass
else:
      __lowerCamelCase						    =      ['''BlenderbotTokenizerFast''']

try:
      if not is_torch_available():
            raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
      pass
else:
      __lowerCamelCase						    =      [
          '''BLENDERBOT_PRETRAINED_MODEL_ARCHIVE_LIST''',
          '''BlenderbotForCausalLM''',
          '''BlenderbotForConditionalGeneration''',
          '''BlenderbotModel''',
          '''BlenderbotPreTrainedModel''',
      ]


try:
      if not is_tf_available():
            raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
      pass
else:
      __lowerCamelCase						    =      [
          '''TFBlenderbotForConditionalGeneration''',
          '''TFBlenderbotModel''',
          '''TFBlenderbotPreTrainedModel''',
      ]


try:
      if not is_flax_available():
            raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
      pass
else:
      __lowerCamelCase						    =      [
          '''FlaxBlenderbotForConditionalGeneration''',
          '''FlaxBlenderbotModel''',
          '''FlaxBlenderbotPreTrainedModel''',
      ]


if TYPE_CHECKING:
      from .configuration_blenderbot import (
          BLENDERBOT_PRETRAINED_CONFIG_ARCHIVE_MAP,
          BlenderbotConfig,
          BlenderbotOnnxConfig,
      )
      from .tokenization_blenderbot import BlenderbotTokenizer

      try:
            if not is_tokenizers_available():
                  raise OptionalDependencyNotAvailable()
      except OptionalDependencyNotAvailable:
            pass
      else:
            from .tokenization_blenderbot_fast import BlenderbotTokenizerFast

      try:
            if not is_torch_available():
                  raise OptionalDependencyNotAvailable()
      except OptionalDependencyNotAvailable:
            pass
      else:
            from .modeling_blenderbot import (
                BLENDERBOT_PRETRAINED_MODEL_ARCHIVE_LIST,
                BlenderbotForCausalLM,
                BlenderbotForConditionalGeneration,
                BlenderbotModel,
                BlenderbotPreTrainedModel,
            )

      try:
            if not is_tf_available():
                  raise OptionalDependencyNotAvailable()
      except OptionalDependencyNotAvailable:
            pass
      else:
            from .modeling_tf_blenderbot import (
                TFBlenderbotForConditionalGeneration,
                TFBlenderbotModel,
                TFBlenderbotPreTrainedModel,
            )

      try:
            if not is_flax_available():
                  raise OptionalDependencyNotAvailable()
      except OptionalDependencyNotAvailable:
            pass
      else:
            from .modeling_flax_blenderbot import (
                FlaxBlenderbotForConditionalGeneration,
                FlaxBlenderbotModel,
                FlaxBlenderbotPreTrainedModel,
            )

else:
      import sys

      __lowerCamelCase						    =      _LazyModule(__name__, globals()['''__file__'''], _import_structure, module_spec=__spec__)
