



"""simple docstring"""
import baseaa



def        lowerCAmelCase_						(							SCREAMING_SNAKE_CASE__	)			->      bytes:
		return baseaa.aaaencode(string.encode("utf-8"	)	)



def        lowerCAmelCase_						(							SCREAMING_SNAKE_CASE__	)			->      str:
		return baseaa.aaadecode(SCREAMING_SNAKE_CASE__	).decode("utf-8"	)


if __name__ == "__main__":
					import doctest

					doctest.testmod()




"""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_      )