



"""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_       (						a_				):
							__lowerCAmelCase    					=					4_2




class 			snake_case_       (						a_     ,a_				):
							__lowerCAmelCase    					=					True






							@register_to_config
							def __init__(   self			,     a_ = 3			,     a_ = 3			,     a_ = ("DownEncoderBlock2D",)			,     a_ = ("UpDecoderBlock2D",)			,     a_ = (6_4,)			,     a_ = 1			,     a_ = "silu"			,     a_ = 4			,     a_ = 3_2			,     a_ = 3_2			,     a_ = 0.18_215			,     ):
									super().__init__()

									# pass init params to Encoder
									a_     :      Optional[Any] 						=     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_     :      Optional[int] 						=     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_     :      Any 						=     nn.Convad(2 * latent_channels			,     2 * latent_channels			,     1      )
									a_     :      int 						=     nn.Convad(a_			,     a_			,     1      )

									a_     :      List[str] 						=     False
									a_     :      List[Any] 						=     False

									# only relevant if vae tiling is enabled
									a_     :      str 						=     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_     :      Dict 						=     int(sample_size / (2 ** (len(self.config.block_out_channels      ) - 1))      )
									a_     :      str 						=     0.25






							def snake_case_  (   self			,     a_			,     a_=False      ):
									if isinstance(a_			,     (Encoder, Decoder)      ):
											a_     :      Tuple 						=     value






							def snake_case_  (   self			,     a_ = True      ):
									a_     :      Dict 						=     use_tiling






							def snake_case_  (   self      ):
									self.enable_tiling(a_      )






							def snake_case_  (   self      ):
									a_     :      Union[str, Any] 						=     True






							def snake_case_  (   self      ):
									a_     :      Any 						=     False






							@property
							# Copied from diffusers.models.unet_2d_condition.UNet2DConditionModel.attn_processors
							def snake_case_  (   self      ):
									a_     :      Tuple 						=     {}

									def fn_recursive_add_processors(a_			,     a_			,     a_      ):
											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 snake_case_  (   self			,     a_      ):
									a_     :      int 						=     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 snake_case_  (   self      ):
									self.set_attn_processor(AttnProcessor()      )






							@apply_forward_hook
							def snake_case_  (   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:
											a_     :      str 						=     [self.encoder(a_      ) for x_slice in x.split(1      )]
											a_     :      int 						=     torch.cat(a_      )
									else:
											a_     :      Tuple 						=     self.encoder(a_      )

									a_     :      str 						=     self.quant_conv(a_      )
									a_     :      Optional[Any] 						=     DiagonalGaussianDistribution(a_      )

									if not return_dict:
											return (posterior,)

									return AutoencoderKLOutput(latent_dist=a_      )






							def snake_case_  (   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_      )

									a_     :      Dict 						=     self.post_quant_conv(a_      )
									a_     :      int 						=     self.decoder(a_      )

									if not return_dict:
											return (dec,)

									return DecoderOutput(sample=a_      )






							@apply_forward_hook
							def snake_case_  (   self			,     a_			,     a_ = True      ):
									if self.use_slicing and z.shape[0] > 1:
											a_     :      Any 						=     [self._decode(a_      ).sample for z_slice in z.split(1      )]
											a_     :      Tuple 						=     torch.cat(a_      )
									else:
											a_     :      Optional[int] 						=     self._decode(a_      ).sample

									if not return_dict:
											return (decoded,)

									return DecoderOutput(sample=a_      )






							def snake_case_  (   self			,     a_			,     a_			,     a_      ):
									a_     :      Dict 						=     min(a.shape[2]			,     b.shape[2]			,     a_      )
									for y in range(a_      ):
											a_     :      Union[str, Any] 						=     a[:, :, -blend_extent + y, :] * (1 - y / blend_extent) + b[:, :, y, :] * (y / blend_extent)
									return b






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






							def snake_case_  (   self			,     a_			,     a_ = True      ):
									a_     :      Optional[int] 						=     int(self.tile_sample_min_size * (1 - self.tile_overlap_factor)      )
									a_     :      List[str] 						=     int(self.tile_latent_min_size * self.tile_overlap_factor      )
									a_     :      Tuple 						=     self.tile_latent_min_size - blend_extent

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

									if not return_dict:
											return (posterior,)

									return AutoencoderKLOutput(latent_dist=a_      )






							def snake_case_  (   self			,     a_			,     a_ = True      ):
									a_     :      str 						=     int(self.tile_latent_min_size * (1 - self.tile_overlap_factor)      )
									a_     :      List[str] 						=     int(self.tile_sample_min_size * self.tile_overlap_factor      )
									a_     :      Any 						=     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_     :      Optional[Any] 						=     []
									for i in range(0			,     z.shape[2]			,     a_      ):
											a_     :      str 						=     []
											for j in range(0			,     z.shape[3]			,     a_      ):
													a_     :      Union[str, Any] 						=     z[:, :, i : i + self.tile_latent_min_size, j : j + self.tile_latent_min_size]
													a_     :      Any 						=     self.post_quant_conv(a_      )
													a_     :      List[str] 						=     self.decoder(a_      )
													row.append(a_      )
											rows.append(a_      )
									a_     :      int 						=     []
									for i, row in enumerate(a_      ):
											a_     :      Tuple 						=     []
											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_     :      List[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_     :      Union[str, Any] 						=     torch.cat(a_			,     dim=2      )
									if not return_dict:
											return (dec,)

									return DecoderOutput(sample=a_      )






							def snake_case_  (   self			,     a_			,     a_ = False			,     a_ = True			,     a_ = None			,     ):
									a_     :      Tuple 						=     sample
									a_     :      List[Any] 						=     self.encode(a_      ).latent_dist
									if sample_posterior:
											a_     :      Dict 						=     posterior.sample(generator=a_      )
									else:
											a_     :      Tuple 						=     posterior.mode()
									a_     :      List[Any] 						=     self.decode(a_      ).sample

									if not return_dict:
											return (dec,)

									return DecoderOutput(sample=a_      )




"""simple docstring"""
from typing import Optional, Tuple, Union

import flax
import flax.linen as nn
import jax
import jax.numpy as jnp
from flax.core.frozen_dict import FrozenDict

from ..configuration_utils import ConfigMixin, flax_register_to_config
from ..utils import BaseOutput
from .embeddings_flax import FlaxTimestepEmbedding, FlaxTimesteps
from .modeling_flax_utils import FlaxModelMixin
from .unet_ad_blocks_flax import (
    FlaxCrossAttnDownBlockaD,
    FlaxDownBlockaD,
    FlaxUNetMidBlockaDCrossAttn,
)




@flax.struct.dataclass
class 			snake_case_						(						a_				):
							__lowerCAmelCase    					=					42
							__lowerCAmelCase    					=					42




class 			snake_case_						(						nn.Module				):
							__lowerCAmelCase    					=					42
							__lowerCAmelCase    					=					(1_6, 3_2, 9_6, 2_5_6)
							__lowerCAmelCase    					=					jnp.floataa






							def snake_case_  (   self      ):
									a_     :      Any 						=     nn.Conv(
									    self.block_out_channels[0]			,     kernel_size=(3, 3)			,     padding=((1, 1), (1, 1))			,     dtype=self.dtype			,     )

									a_     :      Tuple 						=     []
									for i in range(len(self.block_out_channels      ) - 1      ):
											a_     :      Tuple 						=     self.block_out_channels[i]
											a_     :      Any 						=     self.block_out_channels[i + 1]
											a_     :      Dict 						=     nn.Conv(
											    a_			,     kernel_size=(3, 3)			,     padding=((1, 1), (1, 1))			,     dtype=self.dtype			,     )
											blocks.append(a_      )
											a_     :      Optional[int] 						=     nn.Conv(
											    a_			,     kernel_size=(3, 3)			,     strides=(2, 2)			,     padding=((1, 1), (1, 1))			,     dtype=self.dtype			,     )
											blocks.append(a_      )
									a_     :      int 						=     blocks

									a_     :      str 						=     nn.Conv(
									    self.conditioning_embedding_channels			,     kernel_size=(3, 3)			,     padding=((1, 1), (1, 1))			,     kernel_init=nn.initializers.zeros_init()			,     bias_init=nn.initializers.zeros_init()			,     dtype=self.dtype			,     )






							def __call__(   self			,     a_      ):
									a_     :      Dict 						=     self.conv_in(a_      )
									a_     :      Optional[Any] 						=     nn.silu(a_      )

									for block in self.blocks:
											a_     :      Tuple 						=     block(a_      )
											a_     :      Union[str, Any] 						=     nn.silu(a_      )

									a_     :      List[str] 						=     self.conv_out(a_      )

									return embedding




@flax_register_to_config
class 			snake_case_						(						nn.Module     ,a_     ,a_				):
							__lowerCAmelCase    					=					3_2
							__lowerCAmelCase    					=					4
							__lowerCAmelCase    					=					(
							    "CrossAttnDownBlock2D",
							    "CrossAttnDownBlock2D",
							    "CrossAttnDownBlock2D",
							    "DownBlock2D",
							)
							__lowerCAmelCase    					=					False
							__lowerCAmelCase    					=					(3_2_0, 6_4_0, 1_2_8_0, 1_2_8_0)
							__lowerCAmelCase    					=					2
							__lowerCAmelCase    					=					8
							__lowerCAmelCase    					=					None
							__lowerCAmelCase    					=					1_2_8_0
							__lowerCAmelCase    					=					0.0
							__lowerCAmelCase    					=					False
							__lowerCAmelCase    					=					jnp.floataa
							__lowerCAmelCase    					=					True
							__lowerCAmelCase    					=					0
							__lowerCAmelCase    					=					"rgb"
							__lowerCAmelCase    					=					(1_6, 3_2, 9_6, 2_5_6)






							def snake_case_  (   self			,     a_      ):
									# init input tensors
									a_     :      Any 						=     (1, self.in_channels, self.sample_size, self.sample_size)
									a_     :      Tuple 						=     jnp.zeros(a_			,     dtype=jnp.floataa      )
									a_     :      Optional[int] 						=     jnp.ones((1,)			,     dtype=jnp.intaa      )
									a_     :      Any 						=     jnp.zeros((1, 1, self.cross_attention_dim)			,     dtype=jnp.floataa      )
									a_     :      Union[str, Any] 						=     (1, 3, self.sample_size * 8, self.sample_size * 8)
									a_     :      Optional[int] 						=     jnp.zeros(a_			,     dtype=jnp.floataa      )

									a_						,				a_     :      List[Any] 						=     jax.random.split(a_      )
									a_     :      Dict 						=     {"params": params_rng, "dropout": dropout_rng}

									return self.init(a_			,     a_			,     a_			,     a_			,     a_      )["params"]






							def snake_case_  (   self      ):
									a_     :      List[Any] 						=     self.block_out_channels
									a_     :      Any 						=     block_out_channels[0] * 4

									# If `num_attention_heads` is not defined (which is the case for most models)
									# it will default to `attention_head_dim`. This looks weird upon first reading it and it is.
									# The reason for this behavior is to correct for incorrectly named variables that were introduced
									# when this library was created. The incorrect naming was only discovered much later in https://github.com/huggingface/diffusers/issues/2011#issuecomment-1547958131
									# Changing `attention_head_dim` to `num_attention_heads` for 40,000+ configurations is too backwards breaking
									# which is why we correct for the naming here.
									a_     :      Optional[int] 						=     self.num_attention_heads or self.attention_head_dim

									# input
									a_     :      Optional[int] 						=     nn.Conv(
									    block_out_channels[0]			,     kernel_size=(3, 3)			,     strides=(1, 1)			,     padding=((1, 1), (1, 1))			,     dtype=self.dtype			,     )

									# time
									a_     :      Optional[Any] 						=     FlaxTimesteps(
									    block_out_channels[0]			,     flip_sin_to_cos=self.flip_sin_to_cos			,     freq_shift=self.config.freq_shift      )
									a_     :      List[str] 						=     FlaxTimestepEmbedding(a_			,     dtype=self.dtype      )

									a_     :      Optional[int] 						=     FlaxControlNetConditioningEmbedding(
									    conditioning_embedding_channels=block_out_channels[0]			,     block_out_channels=self.conditioning_embedding_out_channels			,     )

									a_     :      int 						=     self.only_cross_attention
									if isinstance(a_			,     a_      ):
											a_     :      Tuple 						=     (only_cross_attention,) * len(self.down_block_types      )

									if isinstance(a_			,     a_      ):
											a_     :      int 						=     (num_attention_heads,) * len(self.down_block_types      )

									# down
									a_     :      Any 						=     []
									a_     :      Optional[Any] 						=     []

									a_     :      List[Any] 						=     block_out_channels[0]

									a_     :      Dict 						=     nn.Conv(
									    a_			,     kernel_size=(1, 1)			,     padding="VALID"			,     kernel_init=nn.initializers.zeros_init()			,     bias_init=nn.initializers.zeros_init()			,     dtype=self.dtype			,     )
									controlnet_down_blocks.append(a_      )

									for i, down_block_type in enumerate(self.down_block_types      ):
											a_     :      int 						=     output_channel
											a_     :      Optional[int] 						=     block_out_channels[i]
											a_     :      str 						=     i == len(a_      ) - 1

											if down_block_type == "CrossAttnDownBlock2D":
													a_     :      int 						=     FlaxCrossAttnDownBlockaD(
													    in_channels=a_			,     out_channels=a_			,     dropout=self.dropout			,     num_layers=self.layers_per_block			,     num_attention_heads=num_attention_heads[i]			,     add_downsample=not is_final_block			,     use_linear_projection=self.use_linear_projection			,     only_cross_attention=only_cross_attention[i]			,     dtype=self.dtype			,     )
											else:
													a_     :      Union[str, Any] 						=     FlaxDownBlockaD(
													    in_channels=a_			,     out_channels=a_			,     dropout=self.dropout			,     num_layers=self.layers_per_block			,     add_downsample=not is_final_block			,     dtype=self.dtype			,     )

											down_blocks.append(a_      )

											for _ in range(self.layers_per_block      ):
													a_     :      Tuple 						=     nn.Conv(
													    a_			,     kernel_size=(1, 1)			,     padding="VALID"			,     kernel_init=nn.initializers.zeros_init()			,     bias_init=nn.initializers.zeros_init()			,     dtype=self.dtype			,     )
													controlnet_down_blocks.append(a_      )

											if not is_final_block:
													a_     :      Optional[Any] 						=     nn.Conv(
													    a_			,     kernel_size=(1, 1)			,     padding="VALID"			,     kernel_init=nn.initializers.zeros_init()			,     bias_init=nn.initializers.zeros_init()			,     dtype=self.dtype			,     )
													controlnet_down_blocks.append(a_      )

									a_     :      Union[str, Any] 						=     down_blocks
									a_     :      Optional[int] 						=     controlnet_down_blocks

									# mid
									a_     :      Union[str, Any] 						=     block_out_channels[-1]
									a_     :      Any 						=     FlaxUNetMidBlockaDCrossAttn(
									    in_channels=a_			,     dropout=self.dropout			,     num_attention_heads=num_attention_heads[-1]			,     use_linear_projection=self.use_linear_projection			,     dtype=self.dtype			,     )

									a_     :      Union[str, Any] 						=     nn.Conv(
									    a_			,     kernel_size=(1, 1)			,     padding="VALID"			,     kernel_init=nn.initializers.zeros_init()			,     bias_init=nn.initializers.zeros_init()			,     dtype=self.dtype			,     )






							def __call__(   self			,     a_			,     a_			,     a_			,     a_			,     a_ = 1.0			,     a_ = True			,     a_ = False			,     ):
									a_     :      int 						=     self.controlnet_conditioning_channel_order
									if channel_order == "bgr":
											a_     :      Tuple 						=     jnp.flip(a_			,     axis=1      )

									# 1. time
									if not isinstance(a_			,     jnp.ndarray      ):
											a_     :      Optional[int] 						=     jnp.array([timesteps]			,     dtype=jnp.intaa      )
									elif isinstance(a_			,     jnp.ndarray      ) and len(timesteps.shape      ) == 0:
											a_     :      Any 						=     timesteps.astype(dtype=jnp.floataa      )
											a_     :      List[Any] 						=     jnp.expand_dims(a_			,     0      )

									a_     :      List[str] 						=     self.time_proj(a_      )
									a_     :      Union[str, Any] 						=     self.time_embedding(a_      )

									# 2. pre-process
									a_     :      Union[str, Any] 						=     jnp.transpose(a_			,     (0, 2, 3, 1)      )
									a_     :      Union[str, Any] 						=     self.conv_in(a_      )

									a_     :      Optional[int] 						=     jnp.transpose(a_			,     (0, 2, 3, 1)      )
									a_     :      int 						=     self.controlnet_cond_embedding(a_      )
									sample += controlnet_cond

									# 3. down
									a_     :      List[str] 						=     (sample,)
									for down_block in self.down_blocks:
											if isinstance(a_			,     a_      ):
													a_						,				a_     :      int 						=     down_block(a_			,     a_			,     a_			,     deterministic=not train      )
											else:
													a_						,				a_     :      Union[str, Any] 						=     down_block(a_			,     a_			,     deterministic=not train      )
											down_block_res_samples += res_samples

									# 4. mid
									a_     :      Optional[Any] 						=     self.mid_block(a_			,     a_			,     a_			,     deterministic=not train      )

									# 5. contronet blocks
									a_     :      Optional[int] 						=     ()
									for down_block_res_sample, controlnet_block in zip(a_			,     self.controlnet_down_blocks      ):
											a_     :      Tuple 						=     controlnet_block(a_      )
											controlnet_down_block_res_samples += (down_block_res_sample,)

									a_     :      Union[str, Any] 						=     controlnet_down_block_res_samples

									a_     :      Optional[int] 						=     self.controlnet_mid_block(a_      )

									# 6. scaling
									a_     :      Optional[Any] 						=     [sample * conditioning_scale for sample in down_block_res_samples]
									mid_block_res_sample *= conditioning_scale

									if not return_dict:
											return (down_block_res_samples, mid_block_res_sample)

									return FlaxControlNetOutput(
									    down_block_res_samples=a_			,     mid_block_res_sample=a_      )