"""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_	(    _a):
      lowerCamelCase__		:    Optional[Any]   =  4_2
      lowerCamelCase__		:    str   =  4_2





class      UpperCAmelCase_	(    nn.Module):
      lowerCamelCase__		:    List[Any]   =  4_2
      lowerCamelCase__		:    Optional[Any]   =  (1_6, 3_2, 9_6, 2_5_6)
      lowerCamelCase__		:    Tuple   =  jnp.floataa


      def 		_UpperCAmelCase       (   self							)       ->					Optional[Any]:
             lowercase__			:		List[Any]       =	nn.Conv(
                 self.block_out_channels[0]		,			kernel_size=(3, 3)		,			padding=((1, 1), (1, 1))		,			dtype=self.dtype		,			)

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

             lowercase__			:		Dict       =	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							)       ->					int:
             lowercase__			:		Union[str, Any]       =	self.conv_in(_snake_case							)
             lowercase__			:		int       =	nn.silu(_snake_case							)

             for block in self.blocks:
                    lowercase__			:		List[str]       =	block(_snake_case							)
                    lowercase__			:		Dict       =	nn.silu(_snake_case							)

             lowercase__			:		List[str]       =	self.conv_out(_snake_case							)

             return embedding





@flax_register_to_config
class      UpperCAmelCase_	(    nn.Module ,							_a ,							_a):
      lowerCamelCase__		:    int   =  3_2
      lowerCamelCase__		:    List[str]   =  4
      lowerCamelCase__		:    List[Any]   =  (
          "CrossAttnDownBlock2D",
          "CrossAttnDownBlock2D",
          "CrossAttnDownBlock2D",
          "DownBlock2D",
      )
      lowerCamelCase__		:    int   =  False
      lowerCamelCase__		:    int   =  (3_2_0, 6_4_0, 1_2_8_0, 1_2_8_0)
      lowerCamelCase__		:    Union[str, Any]   =  2
      lowerCamelCase__		:    Optional[Any]   =  8
      lowerCamelCase__		:    List[str]   =  None
      lowerCamelCase__		:    Any   =  1_2_8_0
      lowerCamelCase__		:    Dict   =  0.0
      lowerCamelCase__		:    Union[str, Any]   =  False
      lowerCamelCase__		:    Union[str, Any]   =  jnp.floataa
      lowerCamelCase__		:    Optional[int]   =  True
      lowerCamelCase__		:    Tuple   =  0
      lowerCamelCase__		:    List[Any]   =  "rgb"
      lowerCamelCase__		:    Optional[int]   =  (1_6, 3_2, 9_6, 2_5_6)


      def 		_UpperCAmelCase       (   self		,			a							)       ->					FrozenDict:
             lowercase__			:		List[Any]       =	(1, self.in_channels, self.sample_size, self.sample_size)
             lowercase__			:		Any       =	jnp.zeros(_snake_case		,			dtype=jnp.floataa							)
             lowercase__			:		Optional[int]       =	jnp.ones((1,)		,			dtype=jnp.intaa							)
             lowercase__			:		Optional[int]       =	jnp.zeros((1, 1, self.cross_attention_dim)		,			dtype=jnp.floataa							)
             lowercase__			:		Any       =	(1, 3, self.sample_size * 8, self.sample_size * 8)
             lowercase__			:		Tuple       =	jnp.zeros(_snake_case		,			dtype=jnp.floataa							)

             lowercase__    ,      lowercase__			:		List[str]       =	jax.random.split(_snake_case							)
             lowercase__			:		int       =	{'params': params_rng, 'dropout': dropout_rng}

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


      def 		_UpperCAmelCase       (   self							)       ->					List[Any]:
             lowercase__			:		List[str]       =	self.block_out_channels
             lowercase__			:		Tuple       =	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.
             lowercase__			:		Any       =	self.num_attention_heads or self.attention_head_dim

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

             # time
             lowercase__			:		Union[str, 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__			:		Union[str, Any]       =	FlaxControlNetConditioningEmbedding(
                 conditioning_embedding_channels=block_out_channels[0]		,			block_out_channels=self.conditioning_embedding_out_channels		,			)

             lowercase__			:		Tuple       =	self.only_cross_attention
             if isinstance(_snake_case		,			_snake_case							):
                    lowercase__			:		Union[str, Any]       =	(only_cross_attention,) * len(self.down_block_types							)

             if isinstance(_snake_case		,			_snake_case							):
                    lowercase__			:		Optional[Any]       =	(num_attention_heads,) * len(self.down_block_types							)

             # down
             lowercase__			:		Optional[Any]       =	[]
             lowercase__			:		Optional[int]       =	[]

             lowercase__			:		Tuple       =	block_out_channels[0]

             lowercase__			:		Optional[int]       =	nn.Conv(
                 _snake_case		,			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(_snake_case							)

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

                    if down_block_type == "CrossAttnDownBlock2D":
                           lowercase__			:		str       =	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]		,			dtype=self.dtype		,			)
                    else:
                           lowercase__			:		Any       =	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							)

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

                    if not is_final_block:
                           lowercase__			:		Any       =	nn.Conv(
                               _snake_case		,			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(_snake_case							)

             lowercase__			:		Optional[Any]       =	down_blocks
             lowercase__			:		Union[str, Any]       =	controlnet_down_blocks

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

             lowercase__			:		Tuple       =	nn.Conv(
                 _snake_case		,			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		,			)       ->					Union[FlaxControlNetOutput, Tuple]:
             lowercase__			:		str       =	self.controlnet_conditioning_channel_order
             if channel_order == "bgr":
                    lowercase__			:		Any       =	jnp.flip(_snake_case		,			axis=1							)

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

             lowercase__			:		int       =	self.time_proj(_snake_case							)
             lowercase__			:		str       =	self.time_embedding(_snake_case							)

             # 2. pre-process
             lowercase__			:		Any       =	jnp.transpose(_snake_case		,			(0, 2, 3, 1)							)
             lowercase__			:		List[Any]       =	self.conv_in(_snake_case							)

             lowercase__			:		Any       =	jnp.transpose(_snake_case		,			(0, 2, 3, 1)							)
             lowercase__			:		Tuple       =	self.controlnet_cond_embedding(_snake_case							)
             sample += controlnet_cond

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

             # 4. mid
             lowercase__			:		List[Any]       =	self.mid_block(_snake_case		,			_snake_case		,			_snake_case		,			deterministic=not train							)

             # 5. contronet blocks
             lowercase__			:		Any       =	()
             for down_block_res_sample, controlnet_block in zip(_snake_case		,			self.controlnet_down_blocks							):
                    lowercase__			:		List[str]       =	controlnet_block(_snake_case							)
                    controlnet_down_block_res_samples += (down_block_res_sample,)

             lowercase__			:		Optional[Any]       =	controlnet_down_block_res_samples

             lowercase__			:		Any       =	self.controlnet_mid_block(_snake_case							)

             # 6. scaling
             lowercase__			:		List[str]       =	[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=_snake_case		,			mid_block_res_sample=_snake_case							)



from typing import Dict

from .base import GenericTensor, Pipeline







class 			__A( a			):




				def 				SCREAMING_SNAKE_CASE_		(     self		,     _snake_case=None		,     _snake_case=None		,     _snake_case=None		,     **_snake_case				)    ->      Optional[Any]:


								'''simple docstring'''




								if tokenize_kwargs is None:
												__a										=    {}

								if truncation is not None:
												if "truncation" in tokenize_kwargs:
																raise ValueError(
																    '''truncation parameter defined twice (given as keyword argument as well as in tokenize_kwargs)'''				)
												__a										=    truncation

								__a										=    tokenize_kwargs

								__a										=    {}
								if return_tensors is not None:
												__a										=    return_tensors

								return preprocess_params, {}, postprocess_params




				def 				SCREAMING_SNAKE_CASE_		(     self		,     _snake_case		,     **_snake_case				)    ->      Dict[str, GenericTensor]:


								'''simple docstring'''




								__a										=    self.framework
								__a										=    self.tokenizer(_snake_case		,     return_tensors=_snake_case		,     **_snake_case				)
								return model_inputs




				def 				SCREAMING_SNAKE_CASE_		(     self		,     _snake_case				)    ->      Optional[Any]:


								'''simple docstring'''




								__a										=    self.model(**_snake_case				)
								return model_outputs




				def 				SCREAMING_SNAKE_CASE_		(     self		,     _snake_case		,     _snake_case=False				)    ->      Optional[int]:


								'''simple docstring'''




								if return_tensors:
												return model_outputs[0]
								if self.framework == "pt":
												return model_outputs[0].tolist()
								elif self.framework == "tf":
												return model_outputs[0].numpy().tolist()




				def __call__(     self		,     *_snake_case		,     **_snake_case				)    ->      Any:


								'''simple docstring'''




								return super().__call__(*_snake_case		,     **_snake_case				)