





"""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      UpperCAmelCase_  (						_snake_case							):
    lowercase__						 =				42
    lowercase__						 =				42






class      UpperCAmelCase_  (						nn.Module							):
    lowercase__						 =				42
    lowercase__						 =				(16, 32, 96, 2_56)
    lowercase__						 =				jnp.floataa




    def 							__magic_name__							(   self						:	Union[str, Any]				)				->							Any:




     '''simple docstring'''

     A__	=    nn.Conv(
         self.block_out_channels[0]     ,			kernel_size=(3, 3)     ,			padding=((1, 1), (1, 1))     ,			dtype=self.dtype     ,			)

     A__	=    []
     for i in range(len(self.block_out_channels				) - 1				):
      A__	=    self.block_out_channels[i]
      A__	=    self.block_out_channels[i + 1]
      A__	=    nn.Conv(
          UpperCamelCase__     ,			kernel_size=(3, 3)     ,			padding=((1, 1), (1, 1))     ,			dtype=self.dtype     ,			)
      blocks.append(UpperCamelCase__				)
      A__	=    nn.Conv(
          UpperCamelCase__     ,			kernel_size=(3, 3)     ,			strides=(2, 2)     ,			padding=((1, 1), (1, 1))     ,			dtype=self.dtype     ,			)
      blocks.append(UpperCamelCase__				)
     A__	=    blocks

     A__	=    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						:	str     ,			snake_case_						:	Tuple				)				->							int:




     '''simple docstring'''

     A__	=    self.conv_in(UpperCamelCase__				)
     A__	=    nn.silu(UpperCamelCase__				)

     for block in self.blocks:
      A__	=    block(UpperCamelCase__				)
      A__	=    nn.silu(UpperCamelCase__				)

     A__	=    self.conv_out(UpperCamelCase__				)

     return embedding






@flax_register_to_config
class      UpperCAmelCase_  (						nn.Module,       _snake_case,       _snake_case							):
    lowercase__						 =				32
    lowercase__						 =				4
    lowercase__						 =				(
        '''CrossAttnDownBlock2D''',
        '''CrossAttnDownBlock2D''',
        '''CrossAttnDownBlock2D''',
        '''DownBlock2D''',
    )
    lowercase__						 =				False
    lowercase__						 =				(3_20, 6_40, 12_80, 12_80)
    lowercase__						 =				2
    lowercase__						 =				8
    lowercase__						 =				None
    lowercase__						 =				12_80
    lowercase__						 =				0.0
    lowercase__						 =				False
    lowercase__						 =				jnp.floataa
    lowercase__						 =				True
    lowercase__						 =				0
    lowercase__						 =				'''rgb'''
    lowercase__						 =				(16, 32, 96, 2_56)




    def 							__magic_name__							(   self						:	Union[str, Any]     ,			snake_case_						:	Any				)				->							FrozenDict:




     '''simple docstring'''

     A__	=    (1, self.in_channels, self.sample_size, self.sample_size)
     A__	=    jnp.zeros(UpperCamelCase__     ,			dtype=jnp.floataa				)
     A__	=    jnp.ones((1,)     ,			dtype=jnp.intaa				)
     A__	=    jnp.zeros((1, 1, self.cross_attention_dim)     ,			dtype=jnp.floataa				)
     A__	=    (1, 3, self.sample_size * 8, self.sample_size * 8)
     A__	=    jnp.zeros(UpperCamelCase__     ,			dtype=jnp.floataa				)

     A__,				A__	=    jax.random.split(UpperCamelCase__				)
     A__	=    {"params": params_rng, "dropout": dropout_rng}

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




    def 							__magic_name__							(   self						:	Union[str, Any]				)				->							List[str]:




     '''simple docstring'''

     A__	=    self.block_out_channels
     A__	=    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__	=    self.num_attention_heads or self.attention_head_dim

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

     # time
     A__	=    FlaxTimesteps(
         block_out_channels[0]     ,			flip_sin_to_cos=self.flip_sin_to_cos     ,			freq_shift=self.config.freq_shift				)
     A__	=    FlaxTimestepEmbedding(UpperCamelCase__     ,			dtype=self.dtype				)

     A__	=    FlaxControlNetConditioningEmbedding(
         conditioning_embedding_channels=block_out_channels[0]     ,			block_out_channels=self.conditioning_embedding_out_channels     ,			)

     A__	=    self.only_cross_attention
     if isinstance(UpperCamelCase__     ,			UpperCamelCase__				):
      A__	=    (only_cross_attention,) * len(self.down_block_types				)

     if isinstance(UpperCamelCase__     ,			UpperCamelCase__				):
      A__	=    (num_attention_heads,) * len(self.down_block_types				)

     # down
     A__	=    []
     A__	=    []

     A__	=    block_out_channels[0]

     A__	=    nn.Conv(
         UpperCamelCase__     ,			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(UpperCamelCase__				)

     for i, down_block_type in enumerate(self.down_block_types				):
      A__	=    output_channel
      A__	=    block_out_channels[i]
      A__	=    i == len(UpperCamelCase__				) - 1

      if down_block_type == "CrossAttnDownBlock2D":
       A__	=    FlaxCrossAttnDownBlockaD(
           in_channels=UpperCamelCase__     ,			out_channels=UpperCamelCase__     ,			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__	=    FlaxDownBlockaD(
           in_channels=UpperCamelCase__     ,			out_channels=UpperCamelCase__     ,			dropout=self.dropout     ,			num_layers=self.layers_per_block     ,			add_downsample=not is_final_block     ,			dtype=self.dtype     ,			)

      down_blocks.append(UpperCamelCase__				)

      for _ in range(self.layers_per_block				):
       A__	=    nn.Conv(
           UpperCamelCase__     ,			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(UpperCamelCase__				)

      if not is_final_block:
       A__	=    nn.Conv(
           UpperCamelCase__     ,			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(UpperCamelCase__				)

     A__	=    down_blocks
     A__	=    controlnet_down_blocks

     # mid
     A__	=    block_out_channels[-1]
     A__	=    FlaxUNetMidBlockaDCrossAttn(
         in_channels=UpperCamelCase__     ,			dropout=self.dropout     ,			num_attention_heads=num_attention_heads[-1]     ,			use_linear_projection=self.use_linear_projection     ,			dtype=self.dtype     ,			)

     A__	=    nn.Conv(
         UpperCamelCase__     ,			kernel_size=(1, 1)     ,			padding="VALID"     ,			kernel_init=nn.initializers.zeros_init()     ,			bias_init=nn.initializers.zeros_init()     ,			dtype=self.dtype     ,			)




    def __call__(   self						:	str     ,			snake_case_						:	str     ,			snake_case_						:	str     ,			snake_case_						:	Tuple     ,			snake_case_						:	Union[str, Any]     ,			snake_case_						:	Optional[Any] = 1.0     ,			snake_case_						:	Optional[Any] = True     ,			snake_case_						:	Union[str, Any] = False     ,			)				->							Union[FlaxControlNetOutput, Tuple]:




     '''simple docstring'''

     A__	=    self.controlnet_conditioning_channel_order
     if channel_order == "bgr":
      A__	=    jnp.flip(UpperCamelCase__     ,			axis=1				)

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

     A__	=    self.time_proj(UpperCamelCase__				)
     A__	=    self.time_embedding(UpperCamelCase__				)

     # 2. pre-process
     A__	=    jnp.transpose(UpperCamelCase__     ,			(0, 2, 3, 1)				)
     A__	=    self.conv_in(UpperCamelCase__				)

     A__	=    jnp.transpose(UpperCamelCase__     ,			(0, 2, 3, 1)				)
     A__	=    self.controlnet_cond_embedding(UpperCamelCase__				)
     sample += controlnet_cond

     # 3. down
     A__	=    (sample,)
     for down_block in self.down_blocks:
      if isinstance(UpperCamelCase__     ,			UpperCamelCase__				):
       A__,				A__	=    down_block(UpperCamelCase__     ,			UpperCamelCase__     ,			UpperCamelCase__     ,			deterministic=not train				)
      else:
       A__,				A__	=    down_block(UpperCamelCase__     ,			UpperCamelCase__     ,			deterministic=not train				)
      down_block_res_samples += res_samples

     # 4. mid
     A__	=    self.mid_block(UpperCamelCase__     ,			UpperCamelCase__     ,			UpperCamelCase__     ,			deterministic=not train				)

     # 5. contronet blocks
     A__	=    ()
     for down_block_res_sample, controlnet_block in zip(UpperCamelCase__     ,			self.controlnet_down_blocks				):
      A__	=    controlnet_block(UpperCamelCase__				)
      controlnet_down_block_res_samples += (down_block_res_sample,)

     A__	=    controlnet_down_block_res_samples

     A__	=    self.controlnet_mid_block(UpperCamelCase__				)

     # 6. scaling
     A__	=    [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=UpperCamelCase__     ,			mid_block_res_sample=UpperCamelCase__				)











"""simple docstring"""



from typing import Optional

from torch import nn

from .transformer_ad import TransformeraDModel, TransformeraDModelOutput




class      UpperCAmelCase_  (						nn.Module							):




    def __init__(   self						:	Optional[int]     ,			snake_case_						:	int = 16     ,			snake_case_						:	int = 88     ,			snake_case_						:	Optional[int] = None     ,			snake_case_						:	int = 1     ,			snake_case_						:	float = 0.0     ,			snake_case_						:	int = 32     ,			snake_case_						:	Optional[int] = None     ,			snake_case_						:	bool = False     ,			snake_case_						:	Optional[int] = None     ,			snake_case_						:	Optional[int] = None     ,			snake_case_						:	str = "geglu"     ,			snake_case_						:	Optional[int] = None     ,			)				->							str:




     '''simple docstring'''

     super().__init__()
     A__	=    nn.ModuleList(
         [
             TransformeraDModel(
                 num_attention_heads=snake_case_     ,			attention_head_dim=snake_case_     ,			in_channels=snake_case_     ,			num_layers=snake_case_     ,			dropout=snake_case_     ,			norm_num_groups=snake_case_     ,			cross_attention_dim=snake_case_     ,			attention_bias=snake_case_     ,			sample_size=snake_case_     ,			num_vector_embeds=snake_case_     ,			activation_fn=snake_case_     ,			num_embeds_ada_norm=snake_case_     ,			)
             for _ in range(2				)
         ]				)

     # Variables that can be set by a pipeline:

     # The ratio of transformer1 to transformer2's output states to be combined during inference
     A__	=    0.5

     # The shape of `encoder_hidden_states` is expected to be
     # `(batch_size, condition_lengths[0]+condition_lengths[1], num_features)`
     A__	=    [77, 257]

     # Which transformer to use to encode which condition.
     # E.g. `(1, 0)` means that we'll use `transformers[1](conditions[0])` and `transformers[0](conditions[1])`
     A__	=    [1, 0]




    def 							__magic_name__							(   self						:	Dict     ,			snake_case_						:	List[Any]     ,			snake_case_						:	Tuple     ,			snake_case_						:	Any=None     ,			snake_case_						:	int=None     ,			snake_case_						:	Union[str, Any]=None     ,			snake_case_						:	bool = True     ,			)				->							Union[str, Any]:




     '''simple docstring'''

     A__	=    hidden_states

     A__	=    []
     A__	=    0
     # attention_mask is not used yet
     for i in range(2				):
      # for each of the two transformers, pass the corresponding condition tokens
      A__	=    encoder_hidden_states[:, tokens_start : tokens_start + self.condition_lengths[i]]
      A__	=    self.transformer_index_for_condition[i]
      A__	=    self.transformers[transformer_index](
          snake_case_     ,			encoder_hidden_states=snake_case_     ,			timestep=snake_case_     ,			cross_attention_kwargs=snake_case_     ,			return_dict=snake_case_     ,			)[0]
      encoded_states.append(encoded_state - input_states				)
      tokens_start += self.condition_lengths[i]

     A__	=    encoded_states[0] * self.mix_ratio + encoded_states[1] * (1 - self.mix_ratio)
     A__	=    output_states + input_states

     if not return_dict:
      return (output_states,)

     return TransformeraDModelOutput(sample=snake_case_				)





