"""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  snake_case__      (       snake_case_       ):
 _snake_case       :					"DiagonalGaussianDistribution"






class  snake_case__      (       snake_case_,						snake_case_       ):
 _snake_case       :					List[Any]							  =						True


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

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

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


 def 						a__   ( self    ,		lowerCamelCase = True						):
        __a       =    use_tiling


 def 						a__   ( self						):
        self.enable_tiling(lowerCamelCase						)


 def 						a__   ( self						):
        __a       =    True


 def 						a__   ( self						):
        __a       =    False


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

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


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

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

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

        if not return_dict:
               return (posterior,)

        return AutoencoderKLOutput(latent_dist=lowerCamelCase						)


 def 						a__   ( self    ,		lowerCamelCase    ,		lowerCamelCase = 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(lowerCamelCase    ,		return_dict=lowerCamelCase						)

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

        if not return_dict:
               return (dec,)

        return DecoderOutput(sample=lowerCamelCase						)


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

        if not return_dict:
               return (decoded,)

        return DecoderOutput(sample=lowerCamelCase						)


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


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


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

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

        if not return_dict:
               return (posterior,)

        return AutoencoderKLOutput(latent_dist=lowerCamelCase						)


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

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

        return DecoderOutput(sample=lowerCamelCase						)






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

        if not return_dict:
               return (dec,)

        return DecoderOutput(sample=lowerCamelCase						)

"""simple docstring"""



import os
from typing import Any, Callable, Dict, List, Optional, Tuple, Union

import torch
from torch import nn

from ...models.controlnet import ControlNetModel, ControlNetOutput
from ...models.modeling_utils import ModelMixin
from ...utils import logging


SCREAMING_SNAKE_CASE__:Optional[Any]									=			logging.get_logger(__name__)






class  snake_case__      (       snake_case_       ):


 def __init__( self    ,		lowerCamelCase						):
        super().__init__()
        __a       =    nn.ModuleList(lowerCamelCase						)


 def 						a__   ( self    ,		lowerCamelCase    ,		lowerCamelCase    ,		lowerCamelCase    ,		lowerCamelCase    ,		lowerCamelCase    ,		lowerCamelCase = None    ,		lowerCamelCase = None    ,		lowerCamelCase = None    ,		lowerCamelCase = None    ,		lowerCamelCase = False    ,		lowerCamelCase = True    ,		):
        for i, (image, scale, controlnet) in enumerate(zip(lowerCamelCase    ,		lowerCamelCase    ,		self.nets						)						):
               __a ,   __a       =    controlnet(
                   lowerCamelCase    ,		lowerCamelCase    ,		lowerCamelCase    ,		lowerCamelCase    ,		lowerCamelCase    ,		lowerCamelCase    ,		lowerCamelCase    ,		lowerCamelCase    ,		lowerCamelCase    ,		lowerCamelCase    ,		lowerCamelCase    ,		)

               # merge samples
               if i == 0:
                      __a ,   __a       =    down_samples, mid_sample
               else:
                      __a       =    [
                          samples_prev + samples_curr
                          for samples_prev, samples_curr in zip(lowerCamelCase    ,		lowerCamelCase						)
                      ]
                      mid_block_res_sample += mid_sample

        return down_block_res_samples, mid_block_res_sample


 def 						a__   ( self    ,		lowerCamelCase    ,		lowerCamelCase = True    ,		lowerCamelCase = None    ,		lowerCamelCase = False    ,		lowerCamelCase = None    ,		):
        __a       =    0
        __a       =    save_directory
        for controlnet in self.nets:
               controlnet.save_pretrained(
                   lowerCamelCase    ,		is_main_process=lowerCamelCase    ,		save_function=lowerCamelCase    ,		safe_serialization=lowerCamelCase    ,		variant=lowerCamelCase    ,		)

               idx += 1
               __a       =    model_path_to_save + F"_{idx}"






 @classmethod
 def 						a__   ( cls    ,		lowerCamelCase    ,		**lowerCamelCase						):
        __a       =    0
        __a       =    []

        # load controlnet and append to list until no controlnet directory exists anymore
        # first controlnet has to be saved under `./mydirectory/controlnet` to be compliant with `DiffusionPipeline.from_prertained`
        # second, third, ... controlnets have to be saved under `./mydirectory/controlnet_1`, `./mydirectory/controlnet_2`, ...
        __a       =    pretrained_model_path
        while os.path.isdir(lowerCamelCase						):
               __a       =    ControlNetModel.from_pretrained(lowerCamelCase    ,		**lowerCamelCase						)
               controlnets.append(lowerCamelCase						)

               idx += 1
               __a       =    pretrained_model_path + F"_{idx}"

        logger.info(F"{len(lowerCamelCase						)} controlnets loaded from {pretrained_model_path}."						)

        if len(lowerCamelCase						) == 0:
               raise ValueError(
                   F"No ControlNets found under {os.path.dirname(lowerCamelCase						)}. Expected at least {pretrained_model_path + '_0'}."						)

        return cls(lowerCamelCase						)
