

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]	      =				(16, 32, 96, 256)
			lowerCamelCase_					:					jnp.dtype	      =				jnp.floataa






			def       lowerCamelCase  (self   )					->       Union[str, Any]:


							'''simple docstring'''



							snake_case_						:   Any           =  nn.Conv(
							    self.block_out_channels[0] ,							kernel_size=(3, 3) ,							padding=((1, 1), (1, 1)) ,							dtype=self.dtype ,							)

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

							snake_case_						:   Optional[Any]           =  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 ,							__magic_name__   )					->       int:


							'''simple docstring'''



							snake_case_						:   int           =  self.conv_in(__magic_name__   )
							snake_case_						:   Union[str, Any]           =  nn.silu(__magic_name__   )

							for block in self.blocks:
											snake_case_						:   Optional[int]           =  block(__magic_name__   )
											snake_case_						:   List[Any]           =  nn.silu(__magic_name__   )

							snake_case_						:   int           =  self.conv_out(__magic_name__   )

							return embedding




@flax_register_to_config
class  __lowerCAmelCase	(      nn.Module,					_a,					_a					):
			lowerCamelCase_					:					int	      =				32
			lowerCamelCase_					:					int	      =				4
			lowerCamelCase_					:					Tuple[str]	      =				(
			    "CrossAttnDownBlock2D",
			    "CrossAttnDownBlock2D",
			    "CrossAttnDownBlock2D",
			    "DownBlock2D",
			)
			lowerCamelCase_					:					Union[bool, Tuple[bool]]	      =				False
			lowerCamelCase_					:					Tuple[int]	      =				(320, 640, 1_280, 1_280)
			lowerCamelCase_					:					int	      =				2
			lowerCamelCase_					:					Union[int, Tuple[int]]	      =				8
			lowerCamelCase_					:					Optional[Union[int, Tuple[int]]]	      =				None
			lowerCamelCase_					:					int	      =				1_280
			lowerCamelCase_					:					float	      =				0.0
			lowerCamelCase_					:					bool	      =				False
			lowerCamelCase_					:					jnp.dtype	      =				jnp.floataa
			lowerCamelCase_					:					bool	      =				True
			lowerCamelCase_					:					int	      =				0
			lowerCamelCase_					:					str	      =				"rgb"
			lowerCamelCase_					:					Tuple[int]	      =				(16, 32, 96, 256)






			def       lowerCamelCase  (self ,							__magic_name__   )					->       FrozenDict:


							'''simple docstring'''



							snake_case_						:   Any           =  (1, self.in_channels, self.sample_size, self.sample_size)
							snake_case_						:   Union[str, Any]           =  jnp.zeros(__magic_name__ ,							dtype=jnp.floataa   )
							snake_case_						:   int           =  jnp.ones((1,) ,							dtype=jnp.intaa   )
							snake_case_						:   Dict           =  jnp.zeros((1, 1, self.cross_attention_dim) ,							dtype=jnp.floataa   )
							snake_case_						:   Any           =  (1, 3, self.sample_size * 8, self.sample_size * 8)
							snake_case_						:   List[str]           =  jnp.zeros(__magic_name__ ,							dtype=jnp.floataa   )

							snake_case_						:   str           =  jax.random.split(__magic_name__   )
							snake_case_						:   Tuple           =  {'params': params_rng, 'dropout': dropout_rng}

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






			def       lowerCamelCase  (self   )					->       List[Any]:


							'''simple docstring'''



							snake_case_						:   List[str]           =  self.block_out_channels
							snake_case_						:   int           =  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.
							snake_case_						:   int           =  self.num_attention_heads or self.attention_head_dim

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

							# time
							snake_case_						:   Tuple           =  FlaxTimesteps(
							    block_out_channels[0] ,							flip_sin_to_cos=self.flip_sin_to_cos ,							freq_shift=self.config.freq_shift   )
							snake_case_						:   int           =  FlaxTimestepEmbedding(__magic_name__ ,							dtype=self.dtype   )

							snake_case_						:   Tuple           =  FlaxControlNetConditioningEmbedding(
							    conditioning_embedding_channels=block_out_channels[0] ,							block_out_channels=self.conditioning_embedding_out_channels ,							)

							snake_case_						:   Optional[Any]           =  self.only_cross_attention
							if isinstance(__magic_name__ ,							__magic_name__   ):
											snake_case_						:   Optional[int]           =  (only_cross_attention,) * len(self.down_block_types   )

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

							# down
							snake_case_						:   List[str]           =  []
							snake_case_						:   List[str]           =  []

							snake_case_						:   List[Any]           =  block_out_channels[0]

							snake_case_						:   Union[str, Any]           =  nn.Conv(
							    __magic_name__ ,							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(__magic_name__   )

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

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

											down_blocks.append(__magic_name__   )

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

											if not is_final_block:
															snake_case_						:   List[str]           =  nn.Conv(
															    __magic_name__ ,							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(__magic_name__   )

							snake_case_						:   Optional[int]           =  down_blocks
							snake_case_						:   Any           =  controlnet_down_blocks

							# mid
							snake_case_						:   int           =  block_out_channels[-1]
							snake_case_						:   Optional[Any]           =  FlaxUNetMidBlockaDCrossAttn(
							    in_channels=__magic_name__ ,							dropout=self.dropout ,							num_attention_heads=num_attention_heads[-1] ,							use_linear_projection=self.use_linear_projection ,							dtype=self.dtype ,							)

							snake_case_						:   Optional[int]           =  nn.Conv(
							    __magic_name__ ,							kernel_size=(1, 1) ,							padding='''VALID''' ,							kernel_init=nn.initializers.zeros_init() ,							bias_init=nn.initializers.zeros_init() ,							dtype=self.dtype ,							)






			def __call__(self ,							__magic_name__ ,							__magic_name__ ,							__magic_name__ ,							__magic_name__ ,							__magic_name__ = 1.0 ,							__magic_name__ = True ,							__magic_name__ = False ,							)					->       Union[FlaxControlNetOutput, Tuple]:


							'''simple docstring'''



							snake_case_						:   List[Any]           =  self.controlnet_conditioning_channel_order
							if channel_order == "bgr":
											snake_case_						:   Tuple           =  jnp.flip(__magic_name__ ,							axis=1   )

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

							snake_case_						:   Optional[int]           =  self.time_proj(__magic_name__   )
							snake_case_						:   Optional[int]           =  self.time_embedding(__magic_name__   )

							# 2. pre-process
							snake_case_						:   Union[str, Any]           =  jnp.transpose(__magic_name__ ,							(0, 2, 3, 1)   )
							snake_case_						:   Dict           =  self.conv_in(__magic_name__   )

							snake_case_						:   Optional[Any]           =  jnp.transpose(__magic_name__ ,							(0, 2, 3, 1)   )
							snake_case_						:   int           =  self.controlnet_cond_embedding(__magic_name__   )
							sample += controlnet_cond

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

							# 4. mid
							snake_case_						:   List[str]           =  self.mid_block(__magic_name__ ,							__magic_name__ ,							__magic_name__ ,							deterministic=not train   )

							# 5. contronet blocks
							snake_case_						:   List[Any]           =  ()
							for down_block_res_sample, controlnet_block in zip(__magic_name__ ,							self.controlnet_down_blocks   ):
											snake_case_						:   Tuple           =  controlnet_block(__magic_name__   )
											controlnet_down_block_res_samples += (down_block_res_sample,)

							snake_case_						:   Optional[Any]           =  controlnet_down_block_res_samples

							snake_case_						:   Union[str, Any]           =  self.controlnet_mid_block(__magic_name__   )

							# 6. scaling
							snake_case_						:   int           =  [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=__magic_name__ ,							mid_block_res_sample=__magic_name__   )







import unittest

import numpy as np
import torch

from diffusers import PNDMPipeline, PNDMScheduler, UNetaDModel
from diffusers.utils.testing_utils import enable_full_determinism, require_torch, slow, torch_device


enable_full_determinism()



class  __lowerCAmelCase	(      unittest.TestCase					):






			@property
			def       lowerCamelCase  (self   )					->       Union[str, Any]:


							'''simple docstring'''



							torch.manual_seed(0   )
							snake_case_						:   int           =  UNetaDModel(
							    block_out_channels=(32, 64) ,							layers_per_block=2 ,							sample_size=32 ,							in_channels=3 ,							out_channels=3 ,							down_block_types=('''DownBlock2D''', '''AttnDownBlock2D''') ,							up_block_types=('''AttnUpBlock2D''', '''UpBlock2D''') ,							)
							return model






			def       lowerCamelCase  (self   )					->       Dict:


							'''simple docstring'''



							snake_case_						:   Any           =  self.dummy_uncond_unet
							snake_case_						:   Optional[Any]           =  PNDMScheduler()

							snake_case_						:   Optional[Any]           =  PNDMPipeline(unet=__magic_name__ ,							scheduler=__magic_name__   )
							pndm.to(__magic_name__   )
							pndm.set_progress_bar_config(disable=__magic_name__   )

							snake_case_						:   str           =  torch.manual_seed(0   )
							snake_case_						:   Dict           =  pndm(generator=__magic_name__ ,							num_inference_steps=20 ,							output_type='''numpy'''   ).images

							snake_case_						:   str           =  torch.manual_seed(0   )
							snake_case_						:   str           =  pndm(generator=__magic_name__ ,							num_inference_steps=20 ,							output_type='''numpy''' ,							return_dict=__magic_name__   )[0]

							snake_case_						:   Any           =  image[0, -3:, -3:, -1]
							snake_case_						:   str           =  image_from_tuple[0, -3:, -3:, -1]

							assert image.shape == (1, 32, 32, 3)
							snake_case_						:   str           =  np.array([1.0, 1.0, 0.0, 1.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       lowerCamelCase  (self   )					->       Any:


							'''simple docstring'''



							snake_case_						:   Tuple           =  '''google/ddpm-cifar10-32'''

							snake_case_						:   Tuple           =  UNetaDModel.from_pretrained(__magic_name__   )
							snake_case_						:   Optional[Any]           =  PNDMScheduler()

							snake_case_						:   Any           =  PNDMPipeline(unet=__magic_name__ ,							scheduler=__magic_name__   )
							pndm.to(__magic_name__   )
							pndm.set_progress_bar_config(disable=__magic_name__   )
							snake_case_						:   int           =  torch.manual_seed(0   )
							snake_case_						:   Tuple           =  pndm(generator=__magic_name__ ,							output_type='''numpy'''   ).images

							snake_case_						:   List[str]           =  image[0, -3:, -3:, -1]

							assert image.shape == (1, 32, 32, 3)
							snake_case_						:   str           =  np.array([0.1_564, 0.14_645, 0.1_406, 0.14_715, 0.12_425, 0.14_045, 0.13_115, 0.12_175, 0.125]   )

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