


"""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      A(  __lowercase		):

		"""simple docstring"""





		A    =							42






@flax_register_to_config
class      A(  nn.Module				,							__lowercase				,							__lowercase		):

		"""simple docstring"""





		A    =							32
		A    =							4
		A    =							4
		A    =							(
		    "CrossAttnDownBlock2D",
		    "CrossAttnDownBlock2D",
		    "CrossAttnDownBlock2D",
		    "DownBlock2D",
		)
		A    =							("UpBlock2D", "CrossAttnUpBlock2D", "CrossAttnUpBlock2D", "CrossAttnUpBlock2D")
		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    =							False
		def   _UpperCamelCase( self					,				SCREAMING_SNAKE_CASE__   )      ->	FrozenDict:

					"""simple docstring"""



					_UpperCamelCase					:Optional[Any]					     =	(1, self.in_channels, self.sample_size, self.sample_size)
					_UpperCamelCase					:Union[str, Any]					     =	jnp.zeros(SCREAMING_SNAKE_CASE__					,				dtype=jnp.floataa   )
					_UpperCamelCase					:Optional[Any]					     =	jnp.ones((1,)					,				dtype=jnp.intaa   )
					_UpperCamelCase					:Dict					     =	jnp.zeros((1, 1, self.cross_attention_dim)					,				dtype=jnp.floataa   )

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

					return self.init(SCREAMING_SNAKE_CASE__					,				SCREAMING_SNAKE_CASE__					,				SCREAMING_SNAKE_CASE__					,				SCREAMING_SNAKE_CASE__   )["params"]
		def   _UpperCamelCase( self   )      ->	Union[str, Any]:

					"""simple docstring"""



					_UpperCamelCase					:str					     =	self.block_out_channels
					_UpperCamelCase					:Union[str, 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.
					_UpperCamelCase					:Any					     =	self.num_attention_heads or self.attention_head_dim

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

					# time
					_UpperCamelCase					:int					     =	FlaxTimesteps(
					    block_out_channels[0]					,				flip_sin_to_cos=self.flip_sin_to_cos					,				freq_shift=self.config.freq_shift   )
					_UpperCamelCase					:List[Any]					     =	FlaxTimestepEmbedding(SCREAMING_SNAKE_CASE__					,				dtype=self.dtype   )

					_UpperCamelCase					:str					     =	self.only_cross_attention
					if isinstance(SCREAMING_SNAKE_CASE__					,				SCREAMING_SNAKE_CASE__   ):
								_UpperCamelCase					:int					     =	(only_cross_attention,) * len(self.down_block_types   )

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

					# down
					_UpperCamelCase					:Optional[int]					     =	[]
					_UpperCamelCase					:str					     =	block_out_channels[0]
					for i, down_block_type in enumerate(self.down_block_types   ):
								_UpperCamelCase					:Dict					     =	output_channel
								_UpperCamelCase					:Dict					     =	block_out_channels[i]
								_UpperCamelCase					:List[Any]					     =	i == len(SCREAMING_SNAKE_CASE__   ) - 1

								if down_block_type == "CrossAttnDownBlock2D":
											_UpperCamelCase					:Optional[int]					     =	FlaxCrossAttnDownBlockaD(
											    in_channels=SCREAMING_SNAKE_CASE__					,				out_channels=SCREAMING_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]					,				use_memory_efficient_attention=self.use_memory_efficient_attention					,				dtype=self.dtype					,				)
								else:
											_UpperCamelCase					:Tuple					     =	FlaxDownBlockaD(
											    in_channels=SCREAMING_SNAKE_CASE__					,				out_channels=SCREAMING_SNAKE_CASE__					,				dropout=self.dropout					,				num_layers=self.layers_per_block					,				add_downsample=not is_final_block					,				dtype=self.dtype					,				)

								down_blocks.append(SCREAMING_SNAKE_CASE__   )
					_UpperCamelCase					:Any					     =	down_blocks

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

								_UpperCamelCase					:int					     =	i == len(SCREAMING_SNAKE_CASE__   ) - 1

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

								up_blocks.append(SCREAMING_SNAKE_CASE__   )
								_UpperCamelCase					:Optional[int]					     =	output_channel
					_UpperCamelCase					:Optional[int]					     =	up_blocks

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





		def __call__( self					,				SCREAMING_SNAKE_CASE__					,				SCREAMING_SNAKE_CASE__					,				SCREAMING_SNAKE_CASE__					,				SCREAMING_SNAKE_CASE__=None					,				SCREAMING_SNAKE_CASE__=None					,				SCREAMING_SNAKE_CASE__ = True					,				SCREAMING_SNAKE_CASE__ = False					,				)      ->	Union[FlaxUNetaDConditionOutput, Tuple]:

					"""simple docstring"""



					if not isinstance(SCREAMING_SNAKE_CASE__					,				jnp.ndarray   ):
								_UpperCamelCase					:Any					     =	jnp.array([timesteps]					,				dtype=jnp.intaa   )
					elif isinstance(SCREAMING_SNAKE_CASE__					,				jnp.ndarray   ) and len(timesteps.shape   ) == 0:
								_UpperCamelCase					:Optional[Any]					     =	timesteps.astype(dtype=jnp.floataa   )
								_UpperCamelCase					:Optional[Any]					     =	jnp.expand_dims(SCREAMING_SNAKE_CASE__					,				0   )

					_UpperCamelCase					:List[Any]					     =	self.time_proj(SCREAMING_SNAKE_CASE__   )
					_UpperCamelCase					:List[str]					     =	self.time_embedding(SCREAMING_SNAKE_CASE__   )

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

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

					if down_block_additional_residuals is not None:
								_UpperCamelCase					:List[Any]					     =	()

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

								_UpperCamelCase					:Tuple					     =	new_down_block_res_samples

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

        # 6. post-process
					_UpperCamelCase					:Tuple					     =	self.conv_norm_out(SCREAMING_SNAKE_CASE__   )
					_UpperCamelCase					:str					     =	nn.silu(SCREAMING_SNAKE_CASE__   )
					_UpperCamelCase					:str					     =	self.conv_out(SCREAMING_SNAKE_CASE__   )
					_UpperCamelCase					:int					     =	jnp.transpose(SCREAMING_SNAKE_CASE__					,				(0, 3, 1, 2)   )

					if not return_dict:
								return (sample,)

					return FlaxUNetaDConditionOutput(sample=SCREAMING_SNAKE_CASE__   )



import torch

from diffusers import KDPMaDiscreteScheduler
from diffusers.utils import torch_device

from .test_schedulers import SchedulerCommonTest







class 							lowerCAmelCase__   ( __lowercase						):
						UpperCamelCase_				:			int						=		(KDPMaDiscreteScheduler,)
						UpperCamelCase_				:			Optional[int]						=		10

						def 			A_					(      self							,				**a					)       ->   int:


								'''simple docstring'''

								_UpperCamelCase         =						{
								    """num_train_timesteps""": 11_00,
								    """beta_start""": 0.0001,
								    """beta_end""": 0.02,
								    """beta_schedule""": """linear""",
								}

								config.update(**a					)
								return config

						def 			A_					(      self					)       ->   List[str]:


								'''simple docstring'''

								for timesteps in [10, 50, 1_00, 10_00]:
										self.check_over_configs(num_train_timesteps=a					)

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


								'''simple docstring'''

								for beta_start, beta_end in zip([0.0_0001, 0.0001, 0.001]							,				[0.0002, 0.002, 0.02]					):
										self.check_over_configs(beta_start=a							,				beta_end=a					)

						def 			A_					(      self					)       ->   Any:


								'''simple docstring'''

								for schedule in ["linear", "scaled_linear"]:
										self.check_over_configs(beta_schedule=a					)

						def 			A_					(      self					)       ->   Optional[int]:


								'''simple docstring'''

								for prediction_type in ["epsilon", "v_prediction"]:
										self.check_over_configs(prediction_type=a					)

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


								'''simple docstring'''

								_UpperCamelCase         =						self.scheduler_classes[0]
								_UpperCamelCase         =						self.get_scheduler_config(prediction_type="""v_prediction"""					)
								_UpperCamelCase         =						scheduler_class(**a					)

								scheduler.set_timesteps(self.num_inference_steps					)

								_UpperCamelCase         =						self.dummy_model()
								_UpperCamelCase         =						self.dummy_sample_deter * scheduler.init_noise_sigma
								_UpperCamelCase         =						sample.to(a					)

								for i, t in enumerate(scheduler.timesteps					):
										_UpperCamelCase         =						scheduler.scale_model_input(a							,				a					)

										_UpperCamelCase         =						model(a							,				a					)

										_UpperCamelCase         =						scheduler.step(a							,				a							,				a					)
										_UpperCamelCase         =						output.prev_sample

								_UpperCamelCase         =						torch.sum(torch.abs(a					)					)
								_UpperCamelCase         =						torch.mean(torch.abs(a					)					)

								if torch_device in ["cpu", "mps"]:
										assert abs(result_sum.item() - 4.6934e-07					) < 1e-2
										assert abs(result_mean.item() - 6.1112e-10					) < 1e-3
								else:
										# CUDA
										assert abs(result_sum.item() - 4.693_4286_5017_0972e-07					) < 1e-2
										assert abs(result_mean.item() - 0.0002					) < 1e-3

						def 			A_					(      self					)       ->   str:


								'''simple docstring'''

								if torch_device == "mps":
										return
								_UpperCamelCase         =						self.scheduler_classes[0]
								_UpperCamelCase         =						self.get_scheduler_config()
								_UpperCamelCase         =						scheduler_class(**a					)

								scheduler.set_timesteps(self.num_inference_steps					)

								_UpperCamelCase         =						self.dummy_model()
								_UpperCamelCase         =						self.dummy_sample_deter * scheduler.init_noise_sigma
								_UpperCamelCase         =						sample.to(a					)

								for i, t in enumerate(scheduler.timesteps					):
										_UpperCamelCase         =						scheduler.scale_model_input(a							,				a					)

										_UpperCamelCase         =						model(a							,				a					)

										_UpperCamelCase         =						scheduler.step(a							,				a							,				a					)
										_UpperCamelCase         =						output.prev_sample

								_UpperCamelCase         =						torch.sum(torch.abs(a					)					)
								_UpperCamelCase         =						torch.mean(torch.abs(a					)					)

								if torch_device in ["cpu", "mps"]:
										assert abs(result_sum.item() - 20.4125					) < 1e-2
										assert abs(result_mean.item() - 0.0266					) < 1e-3
								else:
										# CUDA
										assert abs(result_sum.item() - 20.4125					) < 1e-2
										assert abs(result_mean.item() - 0.0266					) < 1e-3







						def 			A_					(      self					)       ->   str:


								'''simple docstring'''

								if torch_device == "mps":
										return
								_UpperCamelCase         =						self.scheduler_classes[0]
								_UpperCamelCase         =						self.get_scheduler_config()
								_UpperCamelCase         =						scheduler_class(**a					)

								scheduler.set_timesteps(self.num_inference_steps							,				device=a					)

								_UpperCamelCase         =						self.dummy_model()
								_UpperCamelCase         =						self.dummy_sample_deter.to(a					) * scheduler.init_noise_sigma

								for t in scheduler.timesteps:
										_UpperCamelCase         =						scheduler.scale_model_input(a							,				a					)

										_UpperCamelCase         =						model(a							,				a					)

										_UpperCamelCase         =						scheduler.step(a							,				a							,				a					)
										_UpperCamelCase         =						output.prev_sample

								_UpperCamelCase         =						torch.sum(torch.abs(a					)					)
								_UpperCamelCase         =						torch.mean(torch.abs(a					)					)

								if str(a					).startswith("""cpu"""					):
										# The following sum varies between 148 and 156 on mps. Why?
										assert abs(result_sum.item() - 20.4125					) < 1e-2
										assert abs(result_mean.item() - 0.0266					) < 1e-3
								else:
										# CUDA
										assert abs(result_sum.item() - 20.4125					) < 1e-2
										assert abs(result_mean.item() - 0.0266					) < 1e-3



