


'''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 					lowercase_     (    a__   ):
    __UpperCAmelCase   			=	42

class 					lowercase_     (    a__  ,						a__   ):
    __UpperCAmelCase   			=	True


    @register_to_config
    def __init__(     self  ,					a = 3  ,					a = 3  ,					a = ("DownEncoderBlock2D",)  ,					a = ("UpDecoderBlock2D",)  ,					a = (64,)  ,					a = 1  ,					a = "silu"  ,					a = 4  ,					a = 32  ,					a = 32  ,					a = 0.1_8215  ,					):
          super().__init__()

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

          UpperCamelCase__										=				nn.Convad(2 * latent_channels  ,					2 * latent_channels  ,					1    )
          UpperCamelCase__										=				nn.Convad(a  ,					a  ,					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    __a							(     self  ,					a  ,					a=False    ):
          if isinstance(a  ,					(Encoder, Decoder)    ):
                UpperCamelCase__										=				value


    def    __a							(     self  ,					a = True    ):
          UpperCamelCase__										=				use_tiling


    def    __a							(     self    ):
          self.enable_tiling(a    )


    def    __a							(     self    ):
          UpperCamelCase__										=				True


    def    __a							(     self    ):
          UpperCamelCase__										=				False


    @property
    # Copied from diffusers.models.unet_2d_condition.UNet2DConditionModel.attn_processors
    def    __a							(     self    ):
          UpperCamelCase__										=				{}

          def fn_recursive_add_processors(a  ,					a  ,					a    ):
                if hasattr(a  ,					"set_processor"    ):
                      UpperCamelCase__										=				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    __a							(     self  ,					a    ):
          UpperCamelCase__										=				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  ,					a  ,					a    ):
                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    __a							(     self    ):
          self.set_attn_processor(AttnProcessor()    )


    @apply_forward_hook
    def    __a							(     self  ,					a  ,					a = True    ):
          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:
                UpperCamelCase__										=				[self.encoder(a    ) for x_slice in x.split(1    )]
                UpperCamelCase__										=				torch.cat(a    )
          else:
                UpperCamelCase__										=				self.encoder(a    )

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

          if not return_dict:
                return (posterior,)

          return AutoencoderKLOutput(latent_dist=a    )


    def    __a							(     self  ,					a  ,					a = True    ):
          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    )

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

          if not return_dict:
                return (dec,)

          return DecoderOutput(sample=a    )


    @apply_forward_hook
    def    __a							(     self  ,					a  ,					a = True    ):
          if self.use_slicing and z.shape[0] > 1:
                UpperCamelCase__										=				[self._decode(a    ).sample for z_slice in z.split(1    )]
                UpperCamelCase__										=				torch.cat(a    )
          else:
                UpperCamelCase__										=				self._decode(a    ).sample

          if not return_dict:
                return (decoded,)

          return DecoderOutput(sample=a    )


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


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


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

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

          if not return_dict:
                return (posterior,)

          return AutoencoderKLOutput(latent_dist=a    )


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

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

          return DecoderOutput(sample=a    )


    def    __a							(     self  ,					a  ,					a = False  ,					a = True  ,					a = None  ,					):
          UpperCamelCase__										=				sample
          UpperCamelCase__										=				self.encode(a    ).latent_dist
          if sample_posterior:
                UpperCamelCase__										=				posterior.sample(generator=a    )
          else:
                UpperCamelCase__										=				posterior.mode()
          UpperCamelCase__										=				self.decode(a    ).sample

          if not return_dict:
                return (dec,)

          return DecoderOutput(sample=a    )








from typing import Dict

from .base import GenericTensor, Pipeline


class  A						(							_UpperCAmelCase ):







    """simple docstring"""





    def 						snake_case__     (      self :       int,lowercase_ :       Dict=None,lowercase_ :       Tuple=None,lowercase_ :       List[Any]=None,**lowercase_ :       Any		)->    Optional[Any]:





        '''simple docstring'''







        if tokenize_kwargs is None:
            A__          =   {}

        if truncation is not None:
            if "truncation" in tokenize_kwargs:
                raise ValueError(
                    'truncation parameter defined twice (given as keyword argument as well as in tokenize_kwargs)'		)
            A__          =   truncation

        A__          =   tokenize_kwargs

        A__          =   {}
        if return_tensors is not None:
            A__          =   return_tensors

        return preprocess_params, {}, postprocess_params


    def 						snake_case__     (      self :       Dict,lowercase_ :       List[Any],**lowercase_ :       Tuple		)->    Dict[str, GenericTensor]:





        '''simple docstring'''







        A__          =   self.framework
        A__          =   self.tokenizer(lowercase_,return_tensors=lowercase_,**lowercase_		)
        return model_inputs


    def 						snake_case__     (      self :       Tuple,lowercase_ :       int		)->    Optional[Any]:





        '''simple docstring'''







        A__          =   self.model(**lowercase_		)
        return model_outputs


    def 						snake_case__     (      self :       Tuple,lowercase_ :       Tuple,lowercase_ :       List[str]=False		)->    Any:





        '''simple docstring'''







        if return_tensors:
            return model_outputs[0]
        if self.framework == "pt":
            return model_outputs[0].tolist()
        elif self.framework == "tf":
            return model_outputs[0].numpy().tolist()


    def __call__(      self :       List[Any],*lowercase_ :       int,**lowercase_ :       Optional[Any]		)->    int:





        '''simple docstring'''







        return super().__call__(*lowercase_,**lowercase_		)

