

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__	):






		"""simple docstring"""


		a									= 42
		a									= 42






class 	UpperCAmelCase__		(	nn.Module	):






		"""simple docstring"""


		a									= 42
		a									= (16, 32, 96, 2_56)
		a									= jnp.floataa






		def 				lowercase_		(							self						:		Optional[int] )		->							str:
							SCREAMING_SNAKE_CASE__   =   nn.Conv(
							    self.block_out_channels[0]     ,       kernel_size=(3, 3)     ,       padding=((1, 1), (1, 1))     ,       dtype=self.dtype     ,       )

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

							SCREAMING_SNAKE_CASE__   =   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						:		Tuple     ,       __lowerCamelCase						:		List[Any] )		->							Dict:
							SCREAMING_SNAKE_CASE__   =   self.conv_in(lowerCamelCase_ )
							SCREAMING_SNAKE_CASE__   =   nn.silu(lowerCamelCase_ )

							for block in self.blocks:
												SCREAMING_SNAKE_CASE__   =   block(lowerCamelCase_ )
												SCREAMING_SNAKE_CASE__   =   nn.silu(lowerCamelCase_ )

							SCREAMING_SNAKE_CASE__   =   self.conv_out(lowerCamelCase_ )

							return embedding






@flax_register_to_config
class 	UpperCAmelCase__		(	nn.Module						,					A__						,					A__	):






		"""simple docstring"""


		a									= 32
		a									= 4
		a									= (
		    "CrossAttnDownBlock2D",
		    "CrossAttnDownBlock2D",
		    "CrossAttnDownBlock2D",
		    "DownBlock2D",
		)
		a									= False
		a									= (3_20, 6_40, 12_80, 12_80)
		a									= 2
		a									= 8
		a									= None
		a									= 12_80
		a									= 0.0
		a									= False
		a									= jnp.floataa
		a									= True
		a									= 0
		a									= "rgb"
		a									= (16, 32, 96, 2_56)






		def 				lowercase_		(							self						:		List[str]     ,       __lowerCamelCase						:		Optional[Any] )		->							Optional[Any]:
							SCREAMING_SNAKE_CASE__   =   (1, self.in_channels, self.sample_size, self.sample_size)
							SCREAMING_SNAKE_CASE__   =   jnp.zeros(lowerCamelCase_     ,       dtype=jnp.floataa )
							SCREAMING_SNAKE_CASE__   =   jnp.ones((1,)     ,       dtype=jnp.intaa )
							SCREAMING_SNAKE_CASE__   =   jnp.zeros((1, 1, self.cross_attention_dim)     ,       dtype=jnp.floataa )
							SCREAMING_SNAKE_CASE__   =   (1, 3, self.sample_size * 8, self.sample_size * 8)
							SCREAMING_SNAKE_CASE__   =   jnp.zeros(lowerCamelCase_     ,       dtype=jnp.floataa )

							SCREAMING_SNAKE_CASE__,SCREAMING_SNAKE_CASE__   =   jax.random.split(lowerCamelCase_ )
							SCREAMING_SNAKE_CASE__   =   {'''params''': params_rng, '''dropout''': dropout_rng}

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






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

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

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

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

							SCREAMING_SNAKE_CASE__   =   self.only_cross_attention
							if isinstance(lowerCamelCase_     ,       lowerCamelCase_ ):
												SCREAMING_SNAKE_CASE__   =   (only_cross_attention,) * len(self.down_block_types )

							if isinstance(lowerCamelCase_     ,       lowerCamelCase_ ):
												SCREAMING_SNAKE_CASE__   =   (num_attention_heads,) * len(self.down_block_types )

							# down
							SCREAMING_SNAKE_CASE__   =   []
							SCREAMING_SNAKE_CASE__   =   []

							SCREAMING_SNAKE_CASE__   =   block_out_channels[0]

							SCREAMING_SNAKE_CASE__   =   nn.Conv(
							    lowerCamelCase_     ,       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(lowerCamelCase_ )

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

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

												down_blocks.append(lowerCamelCase_ )

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

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

							SCREAMING_SNAKE_CASE__   =   down_blocks
							SCREAMING_SNAKE_CASE__   =   controlnet_down_blocks

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

							SCREAMING_SNAKE_CASE__   =   nn.Conv(
							    lowerCamelCase_     ,       kernel_size=(1, 1)     ,       padding='''VALID'''     ,       kernel_init=nn.initializers.zeros_init()     ,       bias_init=nn.initializers.zeros_init()     ,       dtype=self.dtype     ,       )







		def __call__(							self						:		Dict     ,       __lowerCamelCase						:		str     ,       __lowerCamelCase						:		str     ,       __lowerCamelCase						:		List[str]     ,       __lowerCamelCase						:		Optional[Any]     ,       __lowerCamelCase						:		int = 1.0     ,       __lowerCamelCase						:		Any = True     ,       __lowerCamelCase						:		Optional[int] = False     ,       )		->							Optional[int]:
							SCREAMING_SNAKE_CASE__   =   self.controlnet_conditioning_channel_order
							if channel_order == "bgr":
												SCREAMING_SNAKE_CASE__   =   jnp.flip(lowerCamelCase_     ,       axis=1 )

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

							SCREAMING_SNAKE_CASE__   =   self.time_proj(lowerCamelCase_ )
							SCREAMING_SNAKE_CASE__   =   self.time_embedding(lowerCamelCase_ )

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

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

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

							# 4. mid
							SCREAMING_SNAKE_CASE__   =   self.mid_block(lowerCamelCase_     ,       lowerCamelCase_     ,       lowerCamelCase_     ,       deterministic=not train )

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

							SCREAMING_SNAKE_CASE__   =   controlnet_down_block_res_samples

							SCREAMING_SNAKE_CASE__   =   self.controlnet_mid_block(lowerCamelCase_ )

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





from argparse import ArgumentParser

from .env import EnvironmentCommand
def     UpperCAmelCase_       (						):





     '''simple docstring'''




     SCREAMING_SNAKE_CASE__   =   ArgumentParser('''Diffusers CLI tool'''				,		usage='''diffusers-cli <command> [<args>]'''    )
     SCREAMING_SNAKE_CASE__   =   parser.add_subparsers(help='''diffusers-cli command helpers'''    )

     # Register commands
     EnvironmentCommand.register_subcommand(_A    )

     # Let's go
     SCREAMING_SNAKE_CASE__   =   parser.parse_args()

     if not hasattr(_A				,		'''func'''    ):
          parser.print_help()
          exit(1    )

     # Run
     SCREAMING_SNAKE_CASE__   =   args.func(_A    )
     service.run()


if __name__ == "__main__":
       main()


