



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







@flax.struct.dataclass
class 	_lowerCAmelCase							(			__UpperCAmelCase ):
	__SCREAMING_SNAKE_CASE   :		jnp.ndarray







@flax_register_to_config
class 	_lowerCAmelCase							(			nn.Module      ,			__UpperCAmelCase      ,			__UpperCAmelCase ):
	__SCREAMING_SNAKE_CASE   :		int					=					32
	__SCREAMING_SNAKE_CASE   :		int					=					4
	__SCREAMING_SNAKE_CASE   :		int					=					4
	__SCREAMING_SNAKE_CASE   :		Tuple[str]					=					(
	    "CrossAttnDownBlock2D",
	    "CrossAttnDownBlock2D",
	    "CrossAttnDownBlock2D",
	    "DownBlock2D",
	)
	__SCREAMING_SNAKE_CASE   :		Tuple[str]					=					("UpBlock2D", "CrossAttnUpBlock2D", "CrossAttnUpBlock2D", "CrossAttnUpBlock2D")
	__SCREAMING_SNAKE_CASE   :		Union[bool, Tuple[bool]]					=					False
	__SCREAMING_SNAKE_CASE   :		Tuple[int]					=					(320, 640, 1_280, 1_280)
	__SCREAMING_SNAKE_CASE   :		int					=					2
	__SCREAMING_SNAKE_CASE   :		Union[int, Tuple[int]]					=					8
	__SCREAMING_SNAKE_CASE   :		Optional[Union[int, Tuple[int]]]					=					None
	__SCREAMING_SNAKE_CASE   :		int					=					1_280
	__SCREAMING_SNAKE_CASE   :		float					=					0.0
	__SCREAMING_SNAKE_CASE   :		bool					=					False
	__SCREAMING_SNAKE_CASE   :		jnp.dtype					=					jnp.floataa
	__SCREAMING_SNAKE_CASE   :		bool					=					True
	__SCREAMING_SNAKE_CASE   :		int					=					0
	__SCREAMING_SNAKE_CASE   :		bool					=					False


	def     _a	(self       ,      lowercase							):
				# init input tensors
				A_	:						List[str]	=     (1, self.in_channels, self.sample_size, self.sample_size)
				A_	:						List[str]	=     jnp.zeros(lowercase       ,      dtype=jnp.floataa							)
				A_	:						Union[str, Any]	=     jnp.ones((1,)       ,      dtype=jnp.intaa							)
				A_	:						Any	=     jnp.zeros((1, 1, self.cross_attention_dim)       ,      dtype=jnp.floataa							)

				A_,    A_	:						List[str]	=     jax.random.split(lowercase							)
				A_	:						List[str]	=     {"""params""": params_rng, """dropout""": dropout_rng}

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


	def     _a	(self							):
				A_	:						Dict	=     self.block_out_channels
				A_	:						Optional[Any]	=     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.
				A_	:						int	=     self.num_attention_heads or self.attention_head_dim

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

				# time
				A_	:						Tuple	=     FlaxTimesteps(
				    block_out_channels[0]       ,      flip_sin_to_cos=self.flip_sin_to_cos       ,      freq_shift=self.config.freq_shift							)
				A_	:						Union[str, Any]	=     FlaxTimestepEmbedding(lowercase       ,      dtype=self.dtype							)

				A_	:						List[str]	=     self.only_cross_attention
				if isinstance(lowercase       ,      lowercase							):
							A_	:						Any	=     (only_cross_attention,) * len(self.down_block_types							)

				if isinstance(lowercase       ,      lowercase							):
							A_	:						Tuple	=     (num_attention_heads,) * len(self.down_block_types							)

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

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

							down_blocks.append(lowercase							)
				A_	:						Union[str, Any]	=     down_blocks

				# mid
				A_	:						Optional[Any]	=     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
				A_	:						List[Any]	=     []
				A_	:						List[str]	=     list(reversed(lowercase							)							)
				A_	:						Optional[int]	=     list(reversed(lowercase							)							)
				A_	:						List[Any]	=     list(reversed(lowercase							)							)
				A_	:						Union[str, Any]	=     reversed_block_out_channels[0]
				for i, up_block_type in enumerate(self.up_block_types							):
							A_	:						Union[str, Any]	=     output_channel
							A_	:						Optional[int]	=     reversed_block_out_channels[i]
							A_	:						Tuple	=     reversed_block_out_channels[min(i + 1       ,      len(lowercase							) - 1							)]

							A_	:						Optional[int]	=     i == len(lowercase							) - 1

							if up_block_type == "CrossAttnUpBlock2D":
										A_	:						List[str]	=     FlaxCrossAttnUpBlockaD(
										    in_channels=lowercase       ,      out_channels=lowercase       ,      prev_output_channel=lowercase       ,      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:
										A_	:						Tuple	=     FlaxUpBlockaD(
										    in_channels=lowercase       ,      out_channels=lowercase       ,      prev_output_channel=lowercase       ,      num_layers=self.layers_per_block + 1       ,      add_upsample=not is_final_block       ,      dropout=self.dropout       ,      dtype=self.dtype       ,      )

							up_blocks.append(lowercase							)
							A_	:						List[Any]	=     output_channel
				A_	:						str	=     up_blocks

				# out
				A_	:						Union[str, Any]	=     nn.GroupNorm(num_groups=32       ,      epsilon=1E-5							)
				A_	:						Union[str, Any]	=     nn.Conv(
				    self.out_channels       ,      kernel_size=(3, 3)       ,      strides=(1, 1)       ,      padding=((1, 1), (1, 1))       ,      dtype=self.dtype       ,      )


	def __call__(self       ,      lowercase       ,      lowercase       ,      lowercase       ,      lowercase=None       ,      lowercase=None       ,      lowercase = True       ,      lowercase = False       ,      ):
				# 1. time
				if not isinstance(lowercase       ,      jnp.ndarray							):
							A_	:						Optional[int]	=     jnp.array([timesteps]       ,      dtype=jnp.intaa							)
				elif isinstance(lowercase       ,      jnp.ndarray							) and len(timesteps.shape							) == 0:
							A_	:						Optional[Any]	=     timesteps.astype(dtype=jnp.floataa							)
							A_	:						Union[str, Any]	=     jnp.expand_dims(lowercase       ,      0							)

				A_	:						Tuple	=     self.time_proj(lowercase							)
				A_	:						Union[str, Any]	=     self.time_embedding(lowercase							)

				# 2. pre-process
				A_	:						int	=     jnp.transpose(lowercase       ,      (0, 2, 3, 1)							)
				A_	:						int	=     self.conv_in(lowercase							)

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

				if down_block_additional_residuals is not None:
							A_	:						List[str]	=     ()

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

							A_	:						List[Any]	=     new_down_block_res_samples

				# 4. mid
				A_	:						int	=     self.mid_block(lowercase       ,      lowercase       ,      lowercase       ,      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:
							A_	:						List[Any]	=     down_block_res_samples[-(self.layers_per_block + 1) :]
							A_	:						str	=     down_block_res_samples[: -(self.layers_per_block + 1)]
							if isinstance(lowercase       ,      lowercase							):
										A_	:						str	=     up_block(
										    lowercase       ,      temb=lowercase       ,      encoder_hidden_states=lowercase       ,      res_hidden_states_tuple=lowercase       ,      deterministic=not train       ,      )
							else:
										A_	:						Any	=     up_block(lowercase       ,      temb=lowercase       ,      res_hidden_states_tuple=lowercase       ,      deterministic=not train							)

        # 6. post-process
				A_	:						Tuple	=     self.conv_norm_out(lowercase							)
				A_	:						Tuple	=     nn.silu(lowercase							)
				A_	:						Union[str, Any]	=     self.conv_out(lowercase							)
				A_	:						List[Any]	=     jnp.transpose(lowercase       ,      (0, 3, 1, 2)							)

				if not return_dict:
							return (sample,)

				return FlaxUNetaDConditionOutput(sample=lowercase							)




'''simple docstring'''




import re



def 	a			( lowerCamelCase__    ):
			'''simple docstring'''





			return [char.split() for char in re.split(r"""[^ a-z A-Z 0-9 \s]"""						,      str_    )]



def 	a			( lowerCamelCase__    ):
			'''simple docstring'''





			A_	:						Optional[int]	=     split_input(str_    )
			return "".join(
			    ["""""".join([char.capitalize() for char in sub_str]    ) for sub_str in string_split]    )



def 	a			( lowerCamelCase__						,      lowerCamelCase__						,      lowerCamelCase__    ):
			'''simple docstring'''





			try:
						A_	:						List[Any]	=     split_input(lowerCamelCase__    )
						if upper:
									A_	:						Tuple	=     """""".join(
									    [
									        separator.join([char.upper() for char in sub_str]    )
									        for sub_str in string_split
									    ]    )
						else:
									A_	:						Optional[int]	=     """""".join(
									    [
									        separator.join([char.lower() for char in sub_str]    )
									        for sub_str in string_split
									    ]    )
						return res_str
			except IndexError:
						return "not valid string"



def 	a			( lowerCamelCase__    ):
			'''simple docstring'''





			return to_simple_case(lowerCamelCase__    )



def 	a			( lowerCamelCase__    ):
			'''simple docstring'''





			try:
						A_	:						Tuple	=     to_simple_case(lowerCamelCase__    )
						return res_str[0].lower() + res_str[1:]
			except IndexError:
						return "not valid string"



def 	a			( lowerCamelCase__						,      lowerCamelCase__    ):
			'''simple docstring'''





			return to_complex_case(lowerCamelCase__						,      lowerCamelCase__						,      """_"""    )



def 	a			( lowerCamelCase__						,      lowerCamelCase__    ):
			'''simple docstring'''





			return to_complex_case(lowerCamelCase__						,      lowerCamelCase__						,      """-"""    )


if __name__ == "__main__":
						__import__('''doctest''').testmod()