



'''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 							lowercase__	(							lowercase   ):
	lowercase__  				=			42






@flax_register_to_config
class 							lowercase__	(							nn.Module						,					lowercase						,					lowercase   ):
	lowercase__  				=			32
	lowercase__  				=			4
	lowercase__  				=			4
	lowercase__  				=			(
	    "CrossAttnDownBlock2D",
	    "CrossAttnDownBlock2D",
	    "CrossAttnDownBlock2D",
	    "DownBlock2D",
	)
	lowercase__  				=			("UpBlock2D", "CrossAttnUpBlock2D", "CrossAttnUpBlock2D", "CrossAttnUpBlock2D")
	lowercase__  				=			False
	lowercase__  				=			(3_20, 6_40, 12_80, 12_80)
	lowercase__  				=			2
	lowercase__  				=			8
	lowercase__  				=			None
	lowercase__  				=			12_80
	lowercase__  				=			0.0
	lowercase__  				=			False
	lowercase__  				=			jnp.floataa
	lowercase__  				=			True
	lowercase__  				=			0
	lowercase__  				=			False





	def 	UpperCamelCase_  ( self					:				List[str] ,lowerCamelCase__					:				jax.random.KeyArray ):







							'''simple docstring'''
							# init input tensors
							_UpperCamelCase					:		Any 				=   (1, self.in_channels, self.sample_size, self.sample_size)
							_UpperCamelCase					:		Dict 				=   jnp.zeros(lowerCamelCase__ ,dtype=jnp.floataa )
							_UpperCamelCase					:		str 				=   jnp.ones((1,) ,dtype=jnp.intaa )
							_UpperCamelCase					:		str 				=   jnp.zeros((1, 1, self.cross_attention_dim) ,dtype=jnp.floataa )

							_UpperCamelCase				,			_UpperCamelCase					:		Dict 				=   jax.random.split(lowerCamelCase__ )
							_UpperCamelCase					:		Optional[Any] 				=   {'params': params_rng, 'dropout': dropout_rng}

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





	def 	UpperCamelCase_  ( self					:				Optional[Any] ):







							'''simple docstring'''
							_UpperCamelCase					:		Optional[int] 				=   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					:		Tuple 				=   self.num_attention_heads or self.attention_head_dim

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

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

							_UpperCamelCase					:		Tuple 				=   self.only_cross_attention
							if isinstance(lowerCamelCase__ ,lowerCamelCase__ ):
													_UpperCamelCase					:		Any 				=   (only_cross_attention,) * len(self.down_block_types )

							if isinstance(lowerCamelCase__ ,lowerCamelCase__ ):
													_UpperCamelCase					:		Optional[int] 				=   (num_attention_heads,) * len(self.down_block_types )

							# down
							_UpperCamelCase					:		str 				=   []
							_UpperCamelCase					:		Tuple 				=   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					:		Optional[Any] 				=   i == len(lowerCamelCase__ ) - 1

													if down_block_type == "CrossAttnDownBlock2D":
																			_UpperCamelCase					:		Tuple 				=   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] ,use_memory_efficient_attention=self.use_memory_efficient_attention ,dtype=self.dtype ,)
													else:
																			_UpperCamelCase					:		Union[str, Any] 				=   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__ )
							_UpperCamelCase					:		Any 				=   down_blocks

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

													_UpperCamelCase					:		Optional[int] 				=   i == len(lowerCamelCase__ ) - 1

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

													up_blocks.append(lowerCamelCase__ )
													_UpperCamelCase					:		str 				=   output_channel
							_UpperCamelCase					:		Dict 				=   up_blocks

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





	def __call__( self					:				Dict ,lowerCamelCase__					:				Tuple ,lowerCamelCase__					:				int ,lowerCamelCase__					:				Optional[int] ,lowerCamelCase__					:				Optional[int]=None ,lowerCamelCase__					:				Optional[Any]=None ,lowerCamelCase__					:				bool = True ,lowerCamelCase__					:				bool = False ,):







							'''simple docstring'''
							# 1. time
							if not isinstance(lowerCamelCase__ ,jnp.ndarray ):
													_UpperCamelCase					:		Any 				=   jnp.array([timesteps] ,dtype=jnp.intaa )
							elif isinstance(lowerCamelCase__ ,jnp.ndarray ) and len(timesteps.shape ) == 0:
													_UpperCamelCase					:		Dict 				=   timesteps.astype(dtype=jnp.floataa )
													_UpperCamelCase					:		Optional[int] 				=   jnp.expand_dims(lowerCamelCase__ ,0 )

							_UpperCamelCase					:		List[Any] 				=   self.time_proj(lowerCamelCase__ )
							_UpperCamelCase					:		Tuple 				=   self.time_embedding(lowerCamelCase__ )

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

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

							if down_block_additional_residuals is not None:
													_UpperCamelCase					:		Optional[int] 				=   ()

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

													_UpperCamelCase					:		int 				=   new_down_block_res_samples

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

        # 6. post-process
							_UpperCamelCase					:		Optional[int] 				=   self.conv_norm_out(lowerCamelCase__ )
							_UpperCamelCase					:		Tuple 				=   nn.silu(lowerCamelCase__ )
							_UpperCamelCase					:		int 				=   self.conv_out(lowerCamelCase__ )
							_UpperCamelCase					:		Any 				=   jnp.transpose(lowerCamelCase__ ,(0, 3, 1, 2) )

							if not return_dict:
													return (sample,)

							return FlaxUNetaDConditionOutput(sample=lowerCamelCase__ )







# DISCLAIMER: This code is strongly influenced by https://github.com/pesser/pytorch_diffusion
# and https://github.com/hojonathanho/diffusion
import math
from dataclasses import dataclass
from typing import List, Optional, Tuple, Union

import numpy as np
import torch

from diffusers.configuration_utils import ConfigMixin, register_to_config
from diffusers.schedulers.scheduling_utils import SchedulerMixin
from diffusers.utils import BaseOutput, deprecate



@dataclass
# Copied from diffusers.schedulers.scheduling_ddpm.DDPMSchedulerOutput with DDPM->DDIM
class 	_lowercase							(   lowerCAmelCase	):





	"""simple docstring"""
	__A							=							42
	__A							=							None






def     a(      A       :    Optional[Any] ,     A       :    Any=0.999 ,     A       :    Dict="cosine" ,     )		->       Optional[int]:
					"""simple docstring"""





					if alpha_transform_type == "cosine":

										def alpha_bar_fn(A       :    Optional[int]       ):
															return math.cos((t + 0.008) / 1.008 * math.pi / 2       ) ** 2

					elif alpha_transform_type == "exp":

										def alpha_bar_fn(A       :    Any       ):
															return math.exp(t * -12.0       )

					else:
										raise ValueError(f'''Unsupported alpha_tranform_type: {alpha_transform_type}'''       )

					a					     =   []
					for i in range(A       ):
										a					     =   i / num_diffusion_timesteps
										a					     =   (i + 1) / num_diffusion_timesteps
										betas.append(min(1 - alpha_bar_fn(A       ) / alpha_bar_fn(A       ) ,     A       )       )
					return torch.tensor(A ,     dtype=torch.floataa       )







class 	_lowercase							(   lowerCAmelCase,      lowerCAmelCase	):





	"""simple docstring"""
	__A							=							1




	@register_to_config
	def __init__(self		,       lowerCamelCase_ = 1000		,       lowerCamelCase_ = 0.0001		,       lowerCamelCase_ = 0.02		,       lowerCamelCase_ = "linear"		,       lowerCamelCase_ = None		,       lowerCamelCase_ = True		,       lowerCamelCase_ = True		,       lowerCamelCase_ = 0		,       lowerCamelCase_ = "epsilon"		,       lowerCamelCase_ = 1.0		,       **lowerCamelCase_		,       ):
						"""simple docstring"""

						if kwargs.get("set_alpha_to_one"		,       lowerCamelCase_		) is not None:
											a					     =   (
											    "The `set_alpha_to_one` argument is deprecated. Please use `set_alpha_to_zero` instead."
											)
											deprecate("set_alpha_to_one"		,       "1.0.0"		,       lowerCamelCase_		,       standard_warn=lowerCamelCase_		)
											a					     =   kwargs["set_alpha_to_one"]
						if trained_betas is not None:
											a					     =   torch.tensor(lowerCamelCase_		,       dtype=torch.floataa		)
						elif beta_schedule == "linear":
											a					     =   torch.linspace(lowerCamelCase_		,       lowerCamelCase_		,       lowerCamelCase_		,       dtype=torch.floataa		)
						elif beta_schedule == "scaled_linear":
											# this schedule is very specific to the latent diffusion model.
											a					     =   (
											    torch.linspace(beta_start**0.5		,       beta_end**0.5		,       lowerCamelCase_		,       dtype=torch.floataa		) ** 2
											)
						elif beta_schedule == "squaredcos_cap_v2":
											# Glide cosine schedule
											a					     =   betas_for_alpha_bar(lowerCamelCase_		)
						else:
											raise NotImplementedError(F'''{beta_schedule} does is not implemented for {self.__class__}'''		)

						a					     =   1.0 - self.betas
						a					     =   torch.cumprod(self.alphas		,       dim=0		)

						# At every step in inverted ddim, we are looking into the next alphas_cumprod
						# For the final step, there is no next alphas_cumprod, and the index is out of bounds
						# `set_alpha_to_zero` decides whether we set this parameter simply to zero
						# in this case, self.step() just output the predicted noise
						# or whether we use the final alpha of the "non-previous" one.
						a					     =   torch.tensor(0.0		) if set_alpha_to_zero else self.alphas_cumprod[-1]

						# standard deviation of the initial noise distribution
						a					     =   1.0

						# setable values
						a					     =   None
						a					     =   torch.from_numpy(np.arange(0		,       lowerCamelCase_		).copy().astype(np.intaa		)		)




	def      UpperCamelCase_       (self		,       lowerCamelCase_		,       lowerCamelCase_ = None		):
						"""simple docstring"""

						return sample




	def      UpperCamelCase_       (self		,       lowerCamelCase_		,       lowerCamelCase_ = None		):
						"""simple docstring"""

						if num_inference_steps > self.config.num_train_timesteps:
											raise ValueError(
											    F'''`num_inference_steps`: {num_inference_steps} cannot be larger than `self.config.train_timesteps`:'''
											    F''' {self.config.num_train_timesteps} as the unet model trained with this scheduler can only handle'''
											    F''' maximal {self.config.num_train_timesteps} timesteps.'''		)

						a					     =   num_inference_steps
						a					     =   self.config.num_train_timesteps // self.num_inference_steps
						# creates integer timesteps by multiplying by ratio
						# casting to int to avoid issues when num_inference_step is power of 3
						a					     =   (np.arange(0		,       lowerCamelCase_		) * step_ratio).round().copy().astype(np.intaa		)
						a					     =   torch.from_numpy(lowerCamelCase_		).to(lowerCamelCase_		)
						self.timesteps += self.config.steps_offset




	def      UpperCamelCase_       (self		,       lowerCamelCase_		,       lowerCamelCase_		,       lowerCamelCase_		,       lowerCamelCase_ = 0.0		,       lowerCamelCase_ = False		,       lowerCamelCase_ = None		,       lowerCamelCase_ = True		,       ):
						"""simple docstring"""

						a					     =   timestep + self.config.num_train_timesteps // self.num_inference_steps

						# 2. compute alphas, betas
						# change original implementation to exactly match noise levels for analogous forward process
						a					     =   self.alphas_cumprod[timestep]
						a					     =   (
						    self.alphas_cumprod[prev_timestep]
						    if prev_timestep < self.config.num_train_timesteps
						    else self.final_alpha_cumprod
						)

						a					     =   1 - alpha_prod_t

						# 3. compute predicted original sample from predicted noise also called
						# "predicted x_0" of formula (12) from https://arxiv.org/pdf/2010.02502.pdf
						if self.config.prediction_type == "epsilon":
											a					     =   (sample - beta_prod_t ** 0.5 * model_output) / alpha_prod_t ** 0.5
											a					     =   model_output
						elif self.config.prediction_type == "sample":
											a					     =   model_output
											a					     =   (sample - alpha_prod_t ** 0.5 * pred_original_sample) / beta_prod_t ** 0.5
						elif self.config.prediction_type == "v_prediction":
											a					     =   (alpha_prod_t**0.5) * sample - (beta_prod_t**0.5) * model_output
											a					     =   (alpha_prod_t**0.5) * model_output + (beta_prod_t**0.5) * sample
						else:
											raise ValueError(
											    F'''prediction_type given as {self.config.prediction_type} must be one of `epsilon`, `sample`, or'''
											    " `v_prediction`"		)

						# 4. Clip or threshold "predicted x_0"
						if self.config.clip_sample:
											a					     =   pred_original_sample.clamp(
											    -self.config.clip_sample_range		,       self.config.clip_sample_range		)

						# 5. compute "direction pointing to x_t" of formula (12) from https://arxiv.org/pdf/2010.02502.pdf
						a					     =   (1 - alpha_prod_t_prev) ** 0.5 * pred_epsilon

						# 6. compute x_t without "random noise" of formula (12) from https://arxiv.org/pdf/2010.02502.pdf
						a					     =   alpha_prod_t_prev ** 0.5 * pred_original_sample + pred_sample_direction

						if not return_dict:
											return (prev_sample, pred_original_sample)
						return DDIMSchedulerOutput(prev_sample=lowerCamelCase_		,       pred_original_sample=lowerCamelCase_		)




	def __len__(self		):
						"""simple docstring"""

						return self.config.num_train_timesteps
