


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,
    FlaxCrossAttnUpBlockaD,
    FlaxDownBlockaD,
    FlaxUNetMidBlockaDCrossAttn,
    FlaxUpBlockaD,
)






@flax.struct.dataclass
class _lowerCamelCase(		_a     ):
       lowercase_				:			jnp.ndarray






@flax_register_to_config
class _lowerCamelCase(		nn.Module,				_a,				_a     ):
       lowercase_				:			int									=    32
       lowercase_				:			int									=    4
       lowercase_				:			int									=    4
       lowercase_				:			Tuple[str]									=    (
           "CrossAttnDownBlock2D",
           "CrossAttnDownBlock2D",
           "CrossAttnDownBlock2D",
           "DownBlock2D",
       )
       lowercase_				:			Tuple[str]									=    ("UpBlock2D", "CrossAttnUpBlock2D", "CrossAttnUpBlock2D", "CrossAttnUpBlock2D")
       lowercase_				:			Union[bool, Tuple[bool]]									=    False
       lowercase_				:			Tuple[int]									=    (3_20, 6_40, 12_80, 12_80)
       lowercase_				:			int									=    2
       lowercase_				:			Union[int, Tuple[int]]									=    8
       lowercase_				:			Optional[Union[int, Tuple[int]]]									=    None
       lowercase_				:			int									=    12_80
       lowercase_				:			float									=    0.0
       lowercase_				:			bool									=    False
       lowercase_				:			jnp.dtype									=    jnp.floataa
       lowercase_				:			bool									=    True
       lowercase_				:			int									=    0
       lowercase_				:			bool									=    False
       def     UpperCamelCase      (      self,   lowerCamelCase) ->     FrozenDict:




           """simple docstring"""

           _lowercase :							int   	=	(1, self.in_channels, self.sample_size, self.sample_size)
           _lowercase :							Any   	=	jnp.zeros(snake_case_,   dtype=jnp.floataa)
           _lowercase :							Optional[Any]   	=	jnp.ones((1,),   dtype=jnp.intaa)
           _lowercase :							str   	=	jnp.zeros((1, 1, self.cross_attention_dim),   dtype=jnp.floataa)

           _lowercase	,       _lowercase :							Dict   	=	jax.random.split(snake_case_)
           _lowercase :							Any   	=	{'params': params_rng, 'dropout': dropout_rng}

           return self.init(snake_case_,   snake_case_,   snake_case_,   snake_case_)["params"]
       def     UpperCamelCase      (      self) ->     List[Any]:




           """simple docstring"""

           _lowercase :							str   	=	self.block_out_channels
           _lowercase :							List[str]   	=	block_out_channels[0] * 4

           if self.num_attention_heads is not None:
               raise ValueError(
                   'At the moment it is not possible to define the number of attention heads via `num_attention_heads` because of a naming issue as described in https://github.com/huggingface/diffusers/issues/2011#issuecomment-1547958131. Passing `num_attention_heads` will only be supported in diffusers v0.19.')

           # 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.
           _lowercase :							int   	=	self.num_attention_heads or self.attention_head_dim

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

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

           _lowercase :							Dict   	=	self.only_cross_attention
           if isinstance(snake_case_,   snake_case_):
               _lowercase :							str   	=	(only_cross_attention,) * len(self.down_block_types)

           if isinstance(snake_case_,   snake_case_):
               _lowercase :							List[Any]   	=	(num_attention_heads,) * len(self.down_block_types)

           # down
           _lowercase :							Optional[int]   	=	[]
           _lowercase :							Dict   	=	block_out_channels[0]
           for i, down_block_type in enumerate(self.down_block_types):
               _lowercase :							Optional[Any]   	=	output_channel
               _lowercase :							Union[str, Any]   	=	block_out_channels[i]
               _lowercase :							Optional[Any]   	=	i == len(snake_case_) - 1

               if down_block_type == "CrossAttnDownBlock2D":
                   _lowercase :							Optional[int]   	=	FlaxCrossAttnDownBlockaD(
                       in_channels=snake_case_,   out_channels=snake_case_,   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],   use_memory_efficient_attention=self.use_memory_efficient_attention,   dtype=self.dtype,   )
               else:
                   _lowercase :							List[str]   	=	FlaxDownBlockaD(
                       in_channels=snake_case_,   out_channels=snake_case_,   dropout=self.dropout,   num_layers=self.layers_per_block,   add_downsample=not is_final_block,   dtype=self.dtype,   )

               down_blocks.append(snake_case_)
           _lowercase :							List[Any]   	=	down_blocks

           # mid
           _lowercase :							Tuple   	=	FlaxUNetMidBlockaDCrossAttn(
               in_channels=block_out_channels[-1],   dropout=self.dropout,   num_attention_heads=num_attention_heads[-1],   use_linear_projection=self.use_linear_projection,   use_memory_efficient_attention=self.use_memory_efficient_attention,   dtype=self.dtype,   )

           # up
           _lowercase :							int   	=	[]
           _lowercase :							int   	=	list(reversed(snake_case_))
           _lowercase :							List[Any]   	=	list(reversed(snake_case_))
           _lowercase :							List[str]   	=	list(reversed(snake_case_))
           _lowercase :							Any   	=	reversed_block_out_channels[0]
           for i, up_block_type in enumerate(self.up_block_types):
               _lowercase :							Any   	=	output_channel
               _lowercase :							List[str]   	=	reversed_block_out_channels[i]
               _lowercase :							List[Any]   	=	reversed_block_out_channels[min(i + 1,   len(snake_case_) - 1)]

               _lowercase :							Optional[Any]   	=	i == len(snake_case_) - 1

               if up_block_type == "CrossAttnUpBlock2D":
                   _lowercase :							Tuple   	=	FlaxCrossAttnUpBlockaD(
                       in_channels=snake_case_,   out_channels=snake_case_,   prev_output_channel=snake_case_,   num_layers=self.layers_per_block + 1,   num_attention_heads=reversed_num_attention_heads[i],   add_upsample=not is_final_block,   dropout=self.dropout,   use_linear_projection=self.use_linear_projection,   only_cross_attention=only_cross_attention[i],   use_memory_efficient_attention=self.use_memory_efficient_attention,   dtype=self.dtype,   )
               else:
                   _lowercase :							Optional[int]   	=	FlaxUpBlockaD(
                       in_channels=snake_case_,   out_channels=snake_case_,   prev_output_channel=snake_case_,   num_layers=self.layers_per_block + 1,   add_upsample=not is_final_block,   dropout=self.dropout,   dtype=self.dtype,   )

               up_blocks.append(snake_case_)
               _lowercase :							Any   	=	output_channel
           _lowercase :							List[str]   	=	up_blocks

           # out
           _lowercase :							Tuple   	=	nn.GroupNorm(num_groups=32,   epsilon=1E-5)
           _lowercase :							Dict   	=	nn.Conv(
               self.out_channels,   kernel_size=(3, 3),   strides=(1, 1),   padding=((1, 1), (1, 1)),   dtype=self.dtype,   )




       def __call__(      self,   lowerCamelCase,   lowerCamelCase,   lowerCamelCase,   lowerCamelCase=None,   lowerCamelCase=None,   lowerCamelCase = True,   lowerCamelCase = False,   ) ->     Union[FlaxUNetaDConditionOutput, Tuple]:




           """simple docstring"""

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

           _lowercase :							int   	=	self.time_proj(snake_case_)
           _lowercase :							Union[str, Any]   	=	self.time_embedding(snake_case_)

           # 2. pre-process
           _lowercase :							str   	=	jnp.transpose(snake_case_,   (0, 2, 3, 1))
           _lowercase :							str   	=	self.conv_in(snake_case_)

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

           if down_block_additional_residuals is not None:
               _lowercase :							Any   	=	()

               for down_block_res_sample, down_block_additional_residual in zip(
                   snake_case_,   snake_case_):
                   down_block_res_sample += down_block_additional_residual
                   new_down_block_res_samples += (down_block_res_sample,)

               _lowercase :							List[str]   	=	new_down_block_res_samples

           # 4. mid
           _lowercase :							Any   	=	self.mid_block(snake_case_,   snake_case_,   snake_case_,   deterministic=not train)

           if mid_block_additional_residual is not None:
               sample += mid_block_additional_residual

           # 5. up
           for up_block in self.up_blocks:
               _lowercase :							List[Any]   	=	down_block_res_samples[-(self.layers_per_block + 1) :]
               _lowercase :							List[str]   	=	down_block_res_samples[: -(self.layers_per_block + 1)]
               if isinstance(snake_case_,   snake_case_):
                   _lowercase :							List[str]   	=	up_block(
                       snake_case_,   temb=snake_case_,   encoder_hidden_states=snake_case_,   res_hidden_states_tuple=snake_case_,   deterministic=not train,   )
               else:
                   _lowercase :							List[Any]   	=	up_block(snake_case_,   temb=snake_case_,   res_hidden_states_tuple=snake_case_,   deterministic=not train)

        # 6. post-process
           _lowercase :							Tuple   	=	self.conv_norm_out(snake_case_)
           _lowercase :							Optional[Any]   	=	nn.silu(snake_case_)
           _lowercase :							Tuple   	=	self.conv_out(snake_case_)
           _lowercase :							List[Any]   	=	jnp.transpose(snake_case_,   (0, 3, 1, 2))

           if not return_dict:
               return (sample,)

           return FlaxUNetaDConditionOutput(sample=snake_case_)






def        UpperCamelCase_(   lowerCamelCase_							,				lowerCamelCase_				)					->		bool:
    return numa ^ numa < 0


if __name__ == "__main__":
 import doctest

 doctest.testmod()


