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





  """simple docstring"""

  __UpperCAmelCase				:				"DiagonalGaussianDistribution"







class UpperCAmelCase__	( lowercase__    ,				lowercase__	):





  """simple docstring"""

  __UpperCAmelCase				:				Any   =					True


  @register_to_config
  def __init__(		self      :     Tuple							,_a      :     int = 3							,_a      :     int = 3							,_a      :     Tuple[str] = ("DownEncoderBlock2D",)							,_a      :     Tuple[str] = ("UpDecoderBlock2D",)							,_a      :     Tuple[int] = (64,)							,_a      :     int = 1							,_a      :     str = "silu"							,_a      :     int = 4							,_a      :     int = 32							,_a      :     int = 32							,_a      :     float = 0.1_8215							,):


     '''simple docstring'''






     super().__init__()

     # pass init params to Encoder
     _a				:	int			=   Encoder(
         in_channels=_a							,out_channels=_a							,down_block_types=_a							,block_out_channels=_a							,layers_per_block=_a							,act_fn=_a							,norm_num_groups=_a							,double_z=_a							,)

     # pass init params to Decoder
     _a				:	Any			=   Decoder(
         in_channels=_a							,out_channels=_a							,up_block_types=_a							,block_out_channels=_a							,layers_per_block=_a							,norm_num_groups=_a							,act_fn=_a							,)

     _a				:	Optional[int]			=   nn.Convad(2 * latent_channels							,2 * latent_channels							,1       )
     _a				:	Union[str, Any]			=   nn.Convad(_a							,_a							,1       )

     _a				:	List[str]			=   False
     _a				:	Union[str, Any]			=   False

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


  def   __lowercase		(		self      :     Optional[int]							,_a      :     Dict							,_a      :     Dict=False       ):


     '''simple docstring'''






     if isinstance(_a							,(Encoder, Decoder)       ):
        _a				:	Tuple			=   value


  def   __lowercase		(		self      :     Tuple							,_a      :     bool = True       ):


     '''simple docstring'''






     _a				:	str			=   use_tiling


  def   __lowercase		(		self      :     Dict       ):


     '''simple docstring'''






     self.enable_tiling(_a       )


  def   __lowercase		(		self      :     List[str]       ):


     '''simple docstring'''






     _a				:	List[str]			=   True


  def   __lowercase		(		self      :     Optional[Any]       ):


     '''simple docstring'''






     _a				:	Union[str, Any]			=   False


  @property
  # Copied from diffusers.models.unet_2d_condition.UNet2DConditionModel.attn_processors
  def   __lowercase		(		self      :     Optional[Any]       ):


     '''simple docstring'''






     _a				:	Union[str, Any]			=   {}

     def fn_recursive_add_processors(_a      :     str							,_a      :     torch.nn.Module							,_a      :     Dict[str, AttentionProcessor]       ):
        if hasattr(_a							,'set_processor'       ):
           _a				:	Optional[Any]			=   module.processor

        for sub_name, child in module.named_children():
           fn_recursive_add_processors(F"""{name}.{sub_name}"""							,_a							,_a       )

        return processors

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

     return processors


  def   __lowercase		(		self      :     Optional[int]							,_a      :     Union[AttentionProcessor, Dict[str, AttentionProcessor]]       ):


     '''simple docstring'''






     _a				:	str			=   len(self.attn_processors.keys()       )

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

     def fn_recursive_attn_processor(_a      :     str							,_a      :     torch.nn.Module							,_a      :     Union[str, Any]       ):
        if hasattr(_a							,'set_processor'       ):
           if not isinstance(_a							,_a       ):
              module.set_processor(_a       )
           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}"""							,_a							,_a       )

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


  def   __lowercase		(		self      :     Union[str, Any]       ):


     '''simple docstring'''






     self.set_attn_processor(AttnProcessor()       )


  @apply_forward_hook
  def   __lowercase		(		self      :     Dict							,_a      :     torch.FloatTensor							,_a      :     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(_a							,return_dict=_a       )

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

     _a				:	Any			=   self.quant_conv(_a       )
     _a				:	str			=   DiagonalGaussianDistribution(_a       )

     if not return_dict:
        return (posterior,)

     return AutoencoderKLOutput(latent_dist=_a       )


  def   __lowercase		(		self      :     List[str]							,_a      :     torch.FloatTensor							,_a      :     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(_a							,return_dict=_a       )

     _a				:	Any			=   self.post_quant_conv(_a       )
     _a				:	int			=   self.decoder(_a       )

     if not return_dict:
        return (dec,)

     return DecoderOutput(sample=_a       )


  @apply_forward_hook
  def   __lowercase		(		self      :     Any							,_a      :     torch.FloatTensor							,_a      :     bool = True       ):


     '''simple docstring'''






     if self.use_slicing and z.shape[0] > 1:
        _a				:	List[str]			=   [self._decode(_a       ).sample for z_slice in z.split(1       )]
        _a				:	Union[str, Any]			=   torch.cat(_a       )
     else:
        _a				:	Union[str, Any]			=   self._decode(_a       ).sample

     if not return_dict:
        return (decoded,)

     return DecoderOutput(sample=_a       )


  def   __lowercase		(		self      :     Any							,_a      :     Any							,_a      :     Optional[int]							,_a      :     Any       ):


     '''simple docstring'''






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


  def   __lowercase		(		self      :     Optional[Any]							,_a      :     List[Any]							,_a      :     Union[str, Any]							,_a      :     Tuple       ):


     '''simple docstring'''






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


  def   __lowercase		(		self      :     str							,_a      :     torch.FloatTensor							,_a      :     bool = True       ):


     '''simple docstring'''






     _a				:	int			=   int(self.tile_sample_min_size * (1 - self.tile_overlap_factor)       )
     _a				:	Any			=   int(self.tile_latent_min_size * self.tile_overlap_factor       )
     _a				:	List[Any]			=   self.tile_latent_min_size - blend_extent

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

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

     if not return_dict:
        return (posterior,)

     return AutoencoderKLOutput(latent_dist=_a       )


  def   __lowercase		(		self      :     Optional[int]							,_a      :     torch.FloatTensor							,_a      :     bool = True       ):


     '''simple docstring'''






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

     _a				:	int			=   torch.cat(_a							,dim=2       )
     if not return_dict:
        return (dec,)

     return DecoderOutput(sample=_a       )


  def   __lowercase		(		self      :     Optional[int]							,_a      :     torch.FloatTensor							,_a      :     bool = False							,_a      :     bool = True							,_a      :     Optional[torch.Generator] = None							,):


     '''simple docstring'''






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

     if not return_dict:
        return (dec,)

     return DecoderOutput(sample=_a       )





'''simple docstring'''







import doctest
import logging
import os
import unittest
from pathlib import Path
from typing import List, Union

import transformers
from transformers.testing_utils import require_tf, require_torch, slow


__lowerCAmelCase         =    logging.getLogger()







@unittest.skip('''Temporarily disable the doc tests.'''	)
@require_torch
@require_tf
@slow
class UpperCAmelCase__	( unittest.TestCase	):





  """simple docstring"""

  def   __lowercase		(		self      :     str							,_a      :     Path							,_a      :     Union[str, None] = None							,_a      :     Union[List[str], None] = None							,_a      :     Union[str, List[str], None] = None							,_a      :     bool = True							,):


     '''simple docstring'''






     _a				:	Optional[int]			=   [file for file in os.listdir(_a       ) if os.path.isfile(os.path.join(_a							,_a       )       )]

     if identifier is not None:
        _a				:	List[str]			=   [file for file in files if identifier in file]

     if n_identifier is not None:
        if isinstance(_a							,_a       ):
           for n_ in n_identifier:
              _a				:	Tuple			=   [file for file in files if n_ not in file]
        else:
           _a				:	Optional[Any]			=   [file for file in files if n_identifier not in file]

     _a				:	List[str]			=   ignore_files or []
     ignore_files.append('__init__.py'       )
     _a				:	Tuple			=   [file for file in files if file not in ignore_files]

     for file in files:
        # Open all files
        print('Testing'							,_a       )

        if only_modules:
           _a				:	Any			=   file.split('.'       )[0]
           try:
              _a				:	List[str]			=   getattr(_a							,_a       )
              _a				:	int			=   doctest.DocTestSuite(_a       )
              _a				:	Any			=   unittest.TextTestRunner().run(_a       )
              self.assertIs(len(result.failures       )							,0       )
           except AttributeError:
              logger.info(F"""{module_identifier} is not a module."""       )
        else:
           _a				:	Union[str, Any]			=   doctest.testfile(str('..' / directory / file       )							,optionflags=doctest.ELLIPSIS       )
           self.assertIs(result.failed							,0       )


  def   __lowercase		(		self      :     Any       ):


     '''simple docstring'''






     _a				:	int			=   Path('src/transformers'       )
     _a				:	List[Any]			=   'modeling'
     _a				:	Optional[Any]			=   [
         'modeling_ctrl.py',
         'modeling_tf_ctrl.py',
     ]
     self.analyze_directory(_a							,identifier=_a							,ignore_files=_a       )


  def   __lowercase		(		self      :     Optional[int]       ):


     '''simple docstring'''






     _a				:	Optional[Any]			=   Path('src/transformers'       )
     _a				:	Optional[Any]			=   'tokenization'
     self.analyze_directory(_a							,identifier=_a       )


  def   __lowercase		(		self      :     List[str]       ):


     '''simple docstring'''






     _a				:	Dict			=   Path('src/transformers'       )
     _a				:	str			=   'configuration'
     self.analyze_directory(_a							,identifier=_a       )


  def   __lowercase		(		self      :     Any       ):


     '''simple docstring'''






     _a				:	Tuple			=   Path('src/transformers'       )
     _a				:	List[Any]			=   ['configuration', 'modeling', 'tokenization']
     self.analyze_directory(_a							,n_identifier=_a       )


  def   __lowercase		(		self      :     Tuple       ):


     '''simple docstring'''






     _a				:	List[Any]			=   Path('docs/source'       )
     _a				:	List[str]			=   ['favicon.ico']
     self.analyze_directory(_a							,ignore_files=_a							,only_modules=_a       )



