

"""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    lowerCamelCase						(A__			):
			lowerCamelCase__   :		jnp.ndarray
			lowerCamelCase__   :		jnp.ndarray





class    lowerCamelCase						(nn.Module			):
			lowerCamelCase__   :		int
			lowerCamelCase__   :		Tuple[int]							=							(1_6, 3_2, 9_6, 2_5_6)
			lowerCamelCase__   :		jnp.dtype							=							jnp.floataa

			def 					SCREAMING_SNAKE_CASE				(					self				:						Tuple       )   ->    Optional[Any]:
						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(
									    __UpperCAmelCase    ,				kernel_size=(3, 3)    ,				padding=((1, 1), (1, 1))    ,				dtype=self.dtype    ,				)
									blocks.append(__UpperCAmelCase       )
									SCREAMING_SNAKE_CASE__								=      nn.Conv(
									    __UpperCAmelCase    ,				kernel_size=(3, 3)    ,				strides=(2, 2)    ,				padding=((1, 1), (1, 1))    ,				dtype=self.dtype    ,				)
									blocks.append(__UpperCAmelCase       )
						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				:						List[Any]    ,				__UpperCAmelCase				:						Optional[Any]       )   ->    Tuple:
						SCREAMING_SNAKE_CASE__								=      self.conv_in(__UpperCAmelCase       )
						SCREAMING_SNAKE_CASE__								=      nn.silu(__UpperCAmelCase       )

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

						SCREAMING_SNAKE_CASE__								=      self.conv_out(__UpperCAmelCase       )

						return embedding





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

			def 					SCREAMING_SNAKE_CASE				(					self				:						str    ,				__UpperCAmelCase				:						jax.random.KeyArray       )   ->    FrozenDict:
						# init input tensors
						SCREAMING_SNAKE_CASE__								=      (1, self.in_channels, self.sample_size, self.sample_size)
						SCREAMING_SNAKE_CASE__								=      jnp.zeros(__UpperCAmelCase    ,				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(__UpperCAmelCase    ,				dtype=jnp.floataa       )

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

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

			def 					SCREAMING_SNAKE_CASE				(					self				:						List[Any]       )   ->    Optional[int]:
						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(__UpperCAmelCase    ,				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(__UpperCAmelCase    ,				__UpperCAmelCase       ):
									SCREAMING_SNAKE_CASE__								=      (only_cross_attention,) * len(self.down_block_types       )

						if isinstance(__UpperCAmelCase    ,				__UpperCAmelCase       ):
									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(
						    __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       ):
									SCREAMING_SNAKE_CASE__								=      output_channel
									SCREAMING_SNAKE_CASE__								=      block_out_channels[i]
									SCREAMING_SNAKE_CASE__								=      i == len(__UpperCAmelCase       ) - 1

									if down_block_type == "CrossAttnDownBlock2D":
												SCREAMING_SNAKE_CASE__								=      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:
												SCREAMING_SNAKE_CASE__								=      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       ):
												SCREAMING_SNAKE_CASE__								=      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:
												SCREAMING_SNAKE_CASE__								=      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       )

						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=__UpperCAmelCase    ,				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(
						    __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				:						List[Any]    ,				__UpperCAmelCase				:						Optional[Any]    ,				__UpperCAmelCase				:						Optional[int]    ,				__UpperCAmelCase				:						Union[str, Any]    ,				__UpperCAmelCase				:						Optional[int]    ,				__UpperCAmelCase				:						float = 1.0    ,				__UpperCAmelCase				:						bool = True    ,				__UpperCAmelCase				:						bool = False    ,				)   ->    Union[FlaxControlNetOutput, Tuple]:
						SCREAMING_SNAKE_CASE__								=      self.controlnet_conditioning_channel_order
						if channel_order == "bgr":
									SCREAMING_SNAKE_CASE__								=      jnp.flip(__UpperCAmelCase    ,				axis=1       )

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

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

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

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

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

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

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

						SCREAMING_SNAKE_CASE__								=      controlnet_down_block_res_samples

						SCREAMING_SNAKE_CASE__								=      self.controlnet_mid_block(__UpperCAmelCase       )

						# 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=__UpperCAmelCase    ,				mid_block_res_sample=__UpperCAmelCase       )





"""simple docstring"""




import unittest

import numpy as np
import torch

from diffusers import KarrasVePipeline, KarrasVeScheduler, UNetaDModel
from diffusers.utils.testing_utils import enable_full_determinism, require_torch, slow, torch_device


enable_full_determinism()





class    lowerCamelCase						(unittest.TestCase			):

			@property
			def 					SCREAMING_SNAKE_CASE				(					self				:						Dict       )   ->    Any:
						torch.manual_seed(0       )
						SCREAMING_SNAKE_CASE__								=      UNetaDModel(
						    block_out_channels=(3_2, 6_4)    ,				layers_per_block=2    ,				sample_size=3_2    ,				in_channels=3    ,				out_channels=3    ,				down_block_types=("""DownBlock2D""", """AttnDownBlock2D""")    ,				up_block_types=("""AttnUpBlock2D""", """UpBlock2D""")    ,				)
						return model





			def 					SCREAMING_SNAKE_CASE				(					self				:						Any       )   ->    Tuple:
						SCREAMING_SNAKE_CASE__								=      self.dummy_uncond_unet
						SCREAMING_SNAKE_CASE__								=      KarrasVeScheduler()

						SCREAMING_SNAKE_CASE__								=      KarrasVePipeline(unet=__UpperCAmelCase    ,				scheduler=__UpperCAmelCase       )
						pipe.to(__UpperCAmelCase       )
						pipe.set_progress_bar_config(disable=__UpperCAmelCase       )

						SCREAMING_SNAKE_CASE__								=      torch.manual_seed(0       )
						SCREAMING_SNAKE_CASE__								=      pipe(num_inference_steps=2    ,				generator=__UpperCAmelCase    ,				output_type="""numpy"""       ).images

						SCREAMING_SNAKE_CASE__								=      torch.manual_seed(0       )
						SCREAMING_SNAKE_CASE__								=      pipe(num_inference_steps=2    ,				generator=__UpperCAmelCase    ,				output_type="""numpy"""    ,				return_dict=__UpperCAmelCase       )[0]

						SCREAMING_SNAKE_CASE__								=      image[0, -3:, -3:, -1]
						SCREAMING_SNAKE_CASE__								=      image_from_tuple[0, -3:, -3:, -1]

						assert image.shape == (1, 3_2, 3_2, 3)
						SCREAMING_SNAKE_CASE__								=      np.array([0.0, 1.0, 0.0, 0.0, 0.0, 1.0, 0.0, 0.0, 0.0]       )

						assert np.abs(image_slice.flatten() - expected_slice       ).max() < 1e-2
						assert np.abs(image_from_tuple_slice.flatten() - expected_slice       ).max() < 1e-2





@slow
@require_torch
class    lowerCamelCase						(unittest.TestCase			):

			def 					SCREAMING_SNAKE_CASE				(					self				:						Optional[Any]       )   ->    Optional[int]:
						SCREAMING_SNAKE_CASE__								=      """google/ncsnpp-celebahq-256"""
						SCREAMING_SNAKE_CASE__								=      UNetaDModel.from_pretrained(__UpperCAmelCase       )
						SCREAMING_SNAKE_CASE__								=      KarrasVeScheduler()

						SCREAMING_SNAKE_CASE__								=      KarrasVePipeline(unet=__UpperCAmelCase    ,				scheduler=__UpperCAmelCase       )
						pipe.to(__UpperCAmelCase       )
						pipe.set_progress_bar_config(disable=__UpperCAmelCase       )

						SCREAMING_SNAKE_CASE__								=      torch.manual_seed(0       )
						SCREAMING_SNAKE_CASE__								=      pipe(num_inference_steps=2_0    ,				generator=__UpperCAmelCase    ,				output_type="""numpy"""       ).images

						SCREAMING_SNAKE_CASE__								=      image[0, -3:, -3:, -1]
						assert image.shape == (1, 2_5_6, 2_5_6, 3)
						SCREAMING_SNAKE_CASE__								=      np.array([0.578, 0.5_811, 0.5_924, 0.5_809, 0.587, 0.5_886, 0.5_861, 0.5_802, 0.586]       )

						assert np.abs(image_slice.flatten() - expected_slice       ).max() < 1e-2


