




'''simple docstring'''

import gc
import unittest

import torch
from parameterized import parameterized

from diffusers import AutoencoderKL
from diffusers.utils import floats_tensor, load_hf_numpy, require_torch_gpu, slow, torch_all_close, torch_device
from diffusers.utils.import_utils import is_xformers_available
from diffusers.utils.testing_utils import enable_full_determinism

from .test_modeling_common import ModelTesterMixin, UNetTesterMixin


enable_full_determinism()






class 						a				(   _lowerCamelCase   ,     _lowerCamelCase   ,     unittest.TestCase     ):
				snake_case_      =		AutoencoderKL
				snake_case_      =		"sample"
				snake_case_      =		1e-2


				@property
				def   A_ (		self    :					Dict					):
									snake_case_									=				4
									snake_case_									=				3
									snake_case_									=				(32, 32)

									snake_case_									=				floats_tensor((batch_size, num_channels) + sizes					).to(lowercase_					)

									return {"sample": image}


				@property
				def   A_ (		self    :					List[Any]					):
									return (3, 32, 32)


				@property
				def   A_ (		self    :					Dict					):
									return (3, 32, 32)


				def   A_ (		self    :					Union[str, Any]					):
									snake_case_									=				{
									    '''block_out_channels''': [32, 64],
									    '''in_channels''': 3,
									    '''out_channels''': 3,
									    '''down_block_types''': ['''DownEncoderBlock2D''', '''DownEncoderBlock2D'''],
									    '''up_block_types''': ['''UpDecoderBlock2D''', '''UpDecoderBlock2D'''],
									    '''latent_channels''': 4,
									}
									snake_case_									=				self.dummy_input
									return init_dict, inputs_dict


				def   A_ (		self    :					Any					):
									pass


				def   A_ (		self    :					str					):
									pass


				@unittest.skipIf(torch_device == '''mps'''       ,				'''Gradient checkpointing skipped on MPS'''					)
				def   A_ (		self    :					Dict					):
									# enable deterministic behavior for gradient checkpointing
									snake_case_			,snake_case_									=				self.prepare_init_args_and_inputs_for_common()
									snake_case_									=				self.model_class(**lowercase_					)
									model.to(lowercase_					)

									assert not model.is_gradient_checkpointing and model.training

									snake_case_									=				model(**lowercase_					).sample
									# run the backwards pass on the model. For backwards pass, for simplicity purpose,
									# we won't calculate the loss and rather backprop on out.sum()
									model.zero_grad()

									snake_case_									=				torch.randn_like(lowercase_					)
									snake_case_									=				(out - labels).mean()
									loss.backward()

									# re-instantiate the model now enabling gradient checkpointing
									snake_case_									=				self.model_class(**lowercase_					)
									# clone model
									model_a.load_state_dict(model.state_dict()					)
									model_a.to(lowercase_					)
									model_a.enable_gradient_checkpointing()

									assert model_a.is_gradient_checkpointing and model_a.training

									snake_case_									=				model_a(**lowercase_					).sample
									# run the backwards pass on the model. For backwards pass, for simplicity purpose,
									# we won't calculate the loss and rather backprop on out.sum()
									model_a.zero_grad()
									snake_case_									=				(out_a - labels).mean()
									loss_a.backward()

									# compare the output and parameters gradients
									self.assertTrue((loss - loss_a).abs() < 1e-5					)
									snake_case_									=				dict(model.named_parameters()					)
									snake_case_									=				dict(model_a.named_parameters()					)
									for name, param in named_params.items():
														self.assertTrue(torch_all_close(param.grad.data       ,				named_params_a[name].grad.data       ,				atol=5e-5					)					)


				def   A_ (		self    :					Tuple					):
									snake_case_			,snake_case_									=				AutoencoderKL.from_pretrained('''fusing/autoencoder-kl-dummy'''       ,				output_loading_info=lowercase_					)
									self.assertIsNotNone(lowercase_					)
									self.assertEqual(len(loading_info['''missing_keys''']					)       ,				0					)

									model.to(lowercase_					)
									snake_case_									=				model(**self.dummy_input					)

									assert image is not None, "Make sure output is not None"


				def   A_ (		self    :					Tuple					):
									snake_case_									=				AutoencoderKL.from_pretrained('''fusing/autoencoder-kl-dummy'''					)
									snake_case_									=				model.to(lowercase_					)
									model.eval()

									if torch_device == "mps":
														snake_case_									=				torch.manual_seed(0					)
									else:
														snake_case_									=				torch.Generator(device=lowercase_					).manual_seed(0					)

									snake_case_									=				torch.randn(
									    1       ,				model.config.in_channels       ,				model.config.sample_size       ,				model.config.sample_size       ,				generator=torch.manual_seed(0					)       ,				)
									snake_case_									=				image.to(lowercase_					)
									with torch.no_grad():
														snake_case_									=				model(lowercase_       ,				sample_posterior=lowercase_       ,				generator=lowercase_					).sample

									snake_case_									=				output[0, -1, -3:, -3:].flatten().cpu()

									# Since the VAE Gaussian prior's generator is seeded on the appropriate device,
									# the expected output slices are not the same for CPU and GPU.
									if torch_device == "mps":
														snake_case_									=				torch.tensor(
														    [
														        -4.0_078e-01,
														        -3.8_323e-04,
														        -1.2_681e-01,
														        -1.1_462e-01,
														        2.0_095e-01,
														        1.0_893e-01,
														        -8.8_247e-02,
														        -3.0_361e-01,
														        -9.8_644e-03,
														    ]					)
									elif torch_device == "cpu":
														snake_case_									=				torch.tensor(
														    [-0.1352, 0.0878, 0.0419, -0.0818, -0.1069, 0.0688, -0.1458, -0.4446, -0.0026]					)
									else:
														snake_case_									=				torch.tensor(
														    [-0.2421, 0.4642, 0.2507, -0.0438, 0.0682, 0.3160, -0.2018, -0.0727, 0.2485]					)

									self.assertTrue(torch_all_close(lowercase_       ,				lowercase_       ,				rtol=1e-2					)					)







@slow
class 						a				(   unittest.TestCase     ):


				def   A_ (		self    :					Union[str, Any]       ,				lowercase_    :					Optional[Any]       ,				lowercase_    :					Optional[int]					):
									return F"gaussian_noise_s={seed}_shape={'_'.join([str(lowercase_					) for s in shape]					)}.npy"


				def   A_ (		self    :					Any					):
									# clean up the VRAM after each test
									super().tearDown()
									gc.collect()
									torch.cuda.empty_cache()


				def   A_ (		self    :					Dict       ,				lowercase_    :					List[Any]=0       ,				lowercase_    :					Union[str, Any]=(4, 3, 512, 512)       ,				lowercase_    :					Optional[Any]=False					):
									snake_case_									=				torch.floataa if fpaa else torch.floataa
									snake_case_									=				torch.from_numpy(load_hf_numpy(self.get_file_format(lowercase_       ,				lowercase_					)					)					).to(lowercase_					).to(lowercase_					)
									return image


				def   A_ (		self    :					Any       ,				lowercase_    :					Dict="CompVis/stable-diffusion-v1-4"       ,				lowercase_    :					List[str]=False					):
									snake_case_									=				'''fp16''' if fpaa else None
									snake_case_									=				torch.floataa if fpaa else torch.floataa

									snake_case_									=				AutoencoderKL.from_pretrained(
									    lowercase_       ,				subfolder='''vae'''       ,				torch_dtype=lowercase_       ,				revision=lowercase_       ,				)
									model.to(lowercase_					).eval()

									return model


				def   A_ (		self    :					Any       ,				lowercase_    :					int=0					):
									if torch_device == "mps":
														return torch.manual_seed(lowercase_					)
									return torch.Generator(device=lowercase_					).manual_seed(lowercase_					)


				@parameterized.expand(
				    [
				        # fmt: off
				        [33, [-0.1603, 0.9878, -0.0495, -0.0790, -0.2709, 0.8375, -0.2060, -0.0824], [-0.2395, 0.0098, 0.0102, -0.0709, -0.2840, -0.0274, -0.0718, -0.1824]],
				        [47, [-0.2376, 0.1168, 0.1332, -0.4840, -0.2508, -0.0791, -0.0493, -0.4089], [0.0350, 0.0847, 0.0467, 0.0344, -0.0842, -0.0547, -0.0633, -0.1131]],
				        # fmt: on
				    ]					)
				def   A_ (		self    :					Union[str, Any]       ,				lowercase_    :					Tuple       ,				lowercase_    :					Tuple       ,				lowercase_    :					Tuple					):
									snake_case_									=				self.get_sd_vae_model()
									snake_case_									=				self.get_sd_image(lowercase_					)
									snake_case_									=				self.get_generator(lowercase_					)

									with torch.no_grad():
														snake_case_									=				model(lowercase_       ,				generator=lowercase_       ,				sample_posterior=lowercase_					).sample

									assert sample.shape == image.shape

									snake_case_									=				sample[-1, -2:, -2:, :2].flatten().float().cpu()
									snake_case_									=				torch.tensor(expected_slice_mps if torch_device == '''mps''' else expected_slice					)

									assert torch_all_close(lowercase_       ,				lowercase_       ,				atol=3e-3					)


				@parameterized.expand(
				    [
				        # fmt: off
				        [33, [-0.0513, 0.0289, 1.3799, 0.2166, -0.2573, -0.0871, 0.5103, -0.0999]],
				        [47, [-0.4128, -0.1320, -0.3704, 0.1965, -0.4116, -0.2332, -0.3340, 0.2247]],
				        # fmt: on
				    ]					)
				@require_torch_gpu
				def   A_ (		self    :					Optional[int]       ,				lowercase_    :					Union[str, Any]       ,				lowercase_    :					Dict					):
									snake_case_									=				self.get_sd_vae_model(fpaa=lowercase_					)
									snake_case_									=				self.get_sd_image(lowercase_       ,				fpaa=lowercase_					)
									snake_case_									=				self.get_generator(lowercase_					)

									with torch.no_grad():
														snake_case_									=				model(lowercase_       ,				generator=lowercase_       ,				sample_posterior=lowercase_					).sample

									assert sample.shape == image.shape

									snake_case_									=				sample[-1, -2:, :2, -2:].flatten().float().cpu()
									snake_case_									=				torch.tensor(lowercase_					)

									assert torch_all_close(lowercase_       ,				lowercase_       ,				atol=1e-2					)


				@parameterized.expand(
				    [
				        # fmt: off
				        [33, [-0.1609, 0.9866, -0.0487, -0.0777, -0.2716, 0.8368, -0.2055, -0.0814], [-0.2395, 0.0098, 0.0102, -0.0709, -0.2840, -0.0274, -0.0718, -0.1824]],
				        [47, [-0.2377, 0.1147, 0.1333, -0.4841, -0.2506, -0.0805, -0.0491, -0.4085], [0.0350, 0.0847, 0.0467, 0.0344, -0.0842, -0.0547, -0.0633, -0.1131]],
				        # fmt: on
				    ]					)
				def   A_ (		self    :					Tuple       ,				lowercase_    :					Dict       ,				lowercase_    :					str       ,				lowercase_    :					Optional[int]					):
									snake_case_									=				self.get_sd_vae_model()
									snake_case_									=				self.get_sd_image(lowercase_					)

									with torch.no_grad():
														snake_case_									=				model(lowercase_					).sample

									assert sample.shape == image.shape

									snake_case_									=				sample[-1, -2:, -2:, :2].flatten().float().cpu()
									snake_case_									=				torch.tensor(expected_slice_mps if torch_device == '''mps''' else expected_slice					)

									assert torch_all_close(lowercase_       ,				lowercase_       ,				atol=3e-3					)


				@parameterized.expand(
				    [
				        # fmt: off
				        [13, [-0.2051, -0.1803, -0.2311, -0.2114, -0.3292, -0.3574, -0.2953, -0.3323]],
				        [37, [-0.2632, -0.2625, -0.2199, -0.2741, -0.4539, -0.4990, -0.3720, -0.4925]],
				        # fmt: on
				    ]					)
				@require_torch_gpu
				def   A_ (		self    :					Dict       ,				lowercase_    :					Tuple       ,				lowercase_    :					Optional[int]					):
									snake_case_									=				self.get_sd_vae_model()
									snake_case_									=				self.get_sd_image(lowercase_       ,				shape=(3, 4, 64, 64)					)

									with torch.no_grad():
														snake_case_									=				model.decode(lowercase_					).sample

									assert list(sample.shape					) == [3, 3, 512, 512]

									snake_case_									=				sample[-1, -2:, :2, -2:].flatten().cpu()
									snake_case_									=				torch.tensor(lowercase_					)

									assert torch_all_close(lowercase_       ,				lowercase_       ,				atol=1e-3					)


				@parameterized.expand(
				    [
				        # fmt: off
				        [27, [-0.0369, 0.0207, -0.0776, -0.0682, -0.1747, -0.1930, -0.1465, -0.2039]],
				        [16, [-0.1628, -0.2134, -0.2747, -0.2642, -0.3774, -0.4404, -0.3687, -0.4277]],
				        # fmt: on
				    ]					)
				@require_torch_gpu
				def   A_ (		self    :					Union[str, Any]       ,				lowercase_    :					Any       ,				lowercase_    :					Optional[Any]					):
									snake_case_									=				self.get_sd_vae_model(fpaa=lowercase_					)
									snake_case_									=				self.get_sd_image(lowercase_       ,				shape=(3, 4, 64, 64)       ,				fpaa=lowercase_					)

									with torch.no_grad():
														snake_case_									=				model.decode(lowercase_					).sample

									assert list(sample.shape					) == [3, 3, 512, 512]

									snake_case_									=				sample[-1, -2:, :2, -2:].flatten().float().cpu()
									snake_case_									=				torch.tensor(lowercase_					)

									assert torch_all_close(lowercase_       ,				lowercase_       ,				atol=5e-3					)


				@parameterized.expand([(13,), (16,), (27,)]					)
				@require_torch_gpu
				@unittest.skipIf(not is_xformers_available()       ,				reason='''xformers is not required when using PyTorch 2.0.'''					)
				def   A_ (		self    :					Optional[Any]       ,				lowercase_    :					List[str]					):
									snake_case_									=				self.get_sd_vae_model(fpaa=lowercase_					)
									snake_case_									=				self.get_sd_image(lowercase_       ,				shape=(3, 4, 64, 64)       ,				fpaa=lowercase_					)

									with torch.no_grad():
														snake_case_									=				model.decode(lowercase_					).sample

									model.enable_xformers_memory_efficient_attention()
									with torch.no_grad():
														snake_case_									=				model.decode(lowercase_					).sample

									assert list(sample.shape					) == [3, 3, 512, 512]

									assert torch_all_close(lowercase_       ,				lowercase_       ,				atol=1e-1					)


				@parameterized.expand([(13,), (16,), (37,)]					)
				@require_torch_gpu
				@unittest.skipIf(not is_xformers_available()       ,				reason='''xformers is not required when using PyTorch 2.0.'''					)
				def   A_ (		self    :					Optional[Any]       ,				lowercase_    :					Any					):
									snake_case_									=				self.get_sd_vae_model()
									snake_case_									=				self.get_sd_image(lowercase_       ,				shape=(3, 4, 64, 64)					)

									with torch.no_grad():
														snake_case_									=				model.decode(lowercase_					).sample

									model.enable_xformers_memory_efficient_attention()
									with torch.no_grad():
														snake_case_									=				model.decode(lowercase_					).sample

									assert list(sample.shape					) == [3, 3, 512, 512]

									assert torch_all_close(lowercase_       ,				lowercase_       ,				atol=1e-2					)


				@parameterized.expand(
				    [
				        # fmt: off
				        [33, [-0.3001, 0.0918, -2.6984, -3.9720, -3.2099, -5.0353, 1.7338, -0.2065, 3.4267]],
				        [47, [-1.5030, -4.3871, -6.0355, -9.1157, -1.6661, -2.7853, 2.1607, -5.0823, 2.5633]],
				        # fmt: on
				    ]					)
				def   A_ (		self    :					str       ,				lowercase_    :					Optional[int]       ,				lowercase_    :					Tuple					):
									snake_case_									=				self.get_sd_vae_model()
									snake_case_									=				self.get_sd_image(lowercase_					)
									snake_case_									=				self.get_generator(lowercase_					)

									with torch.no_grad():
														snake_case_									=				model.encode(lowercase_					).latent_dist
														snake_case_									=				dist.sample(generator=lowercase_					)

									assert list(sample.shape					) == [image.shape[0], 4] + [i // 8 for i in image.shape[2:]]

									snake_case_									=				sample[0, -1, -3:, -3:].flatten().cpu()
									snake_case_									=				torch.tensor(lowercase_					)

									snake_case_									=				3e-3 if torch_device != '''mps''' else 1e-2
									assert torch_all_close(lowercase_       ,				lowercase_       ,				atol=lowercase_					)







'''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				(   _lowerCamelCase     ):
				snake_case_      =		42







@flax_register_to_config
class 						a				(   nn.Module   ,     _lowerCamelCase   ,     _lowerCamelCase     ):
				snake_case_      =		32
				snake_case_      =		4
				snake_case_      =		4
				snake_case_      =		(
				    "CrossAttnDownBlock2D",
				    "CrossAttnDownBlock2D",
				    "CrossAttnDownBlock2D",
				    "DownBlock2D",
				)
				snake_case_      =		("UpBlock2D", "CrossAttnUpBlock2D", "CrossAttnUpBlock2D", "CrossAttnUpBlock2D")
				snake_case_      =		False
				snake_case_      =		(320, 640, 1_280, 1_280)
				snake_case_      =		2
				snake_case_      =		8
				snake_case_      =		None
				snake_case_      =		1_280
				snake_case_      =		0.0
				snake_case_      =		False
				snake_case_      =		jnp.floataa
				snake_case_      =		True
				snake_case_      =		0
				snake_case_      =		False


				def   A_ (		self    :					Optional[int]       ,				lowercase_    :					jax.random.KeyArray					):
									# init input tensors
									snake_case_									=				(1, self.in_channels, self.sample_size, self.sample_size)
									snake_case_									=				jnp.zeros(lowercase_       ,				dtype=jnp.floataa					)
									snake_case_									=				jnp.ones((1,)       ,				dtype=jnp.intaa					)
									snake_case_									=				jnp.zeros((1, 1, self.cross_attention_dim)       ,				dtype=jnp.floataa					)

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

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


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

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

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

									snake_case_									=				self.only_cross_attention
									if isinstance(lowercase_       ,				lowercase_					):
														snake_case_									=				(only_cross_attention,) * len(self.down_block_types					)

									if isinstance(lowercase_       ,				lowercase_					):
														snake_case_									=				(num_attention_heads,) * len(self.down_block_types					)

									# down
									snake_case_									=				[]
									snake_case_									=				block_out_channels[0]
									for i, down_block_type in enumerate(self.down_block_types					):
														snake_case_									=				output_channel
														snake_case_									=				block_out_channels[i]
														snake_case_									=				i == len(lowercase_					) - 1

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

														down_blocks.append(lowercase_					)
									snake_case_									=				down_blocks

									# mid
									snake_case_									=				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
									snake_case_									=				[]
									snake_case_									=				list(reversed(lowercase_					)					)
									snake_case_									=				list(reversed(lowercase_					)					)
									snake_case_									=				list(reversed(lowercase_					)					)
									snake_case_									=				reversed_block_out_channels[0]
									for i, up_block_type in enumerate(self.up_block_types					):
														snake_case_									=				output_channel
														snake_case_									=				reversed_block_out_channels[i]
														snake_case_									=				reversed_block_out_channels[min(i + 1       ,				len(lowercase_					) - 1					)]

														snake_case_									=				i == len(lowercase_					) - 1

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

														up_blocks.append(lowercase_					)
														snake_case_									=				output_channel
									snake_case_									=				up_blocks

									# out
									snake_case_									=				nn.GroupNorm(num_groups=32       ,				epsilon=1e-5					)
									snake_case_									=				nn.Conv(
									    self.out_channels       ,				kernel_size=(3, 3)       ,				strides=(1, 1)       ,				padding=((1, 1), (1, 1))       ,				dtype=self.dtype       ,				)


				def __call__(		self    :					Union[str, Any]       ,				lowercase_    :					Union[str, Any]       ,				lowercase_    :					int       ,				lowercase_    :					Any       ,				lowercase_    :					int=None       ,				lowercase_    :					Any=None       ,				lowercase_    :					bool = True       ,				lowercase_    :					bool = False       ,				):
									# 1. time
									if not isinstance(lowercase_       ,				jnp.ndarray					):
														snake_case_									=				jnp.array([timesteps]       ,				dtype=jnp.intaa					)
									elif isinstance(lowercase_       ,				jnp.ndarray					) and len(timesteps.shape					) == 0:
														snake_case_									=				timesteps.astype(dtype=jnp.floataa					)
														snake_case_									=				jnp.expand_dims(lowercase_       ,				0					)

									snake_case_									=				self.time_proj(lowercase_					)
									snake_case_									=				self.time_embedding(lowercase_					)

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

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

									if down_block_additional_residuals is not None:
														snake_case_									=				()

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

														snake_case_									=				new_down_block_res_samples

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

        # 6. post-process
									snake_case_									=				self.conv_norm_out(lowercase_					)
									snake_case_									=				nn.silu(lowercase_					)
									snake_case_									=				self.conv_out(lowercase_					)
									snake_case_									=				jnp.transpose(lowercase_       ,				(0, 3, 1, 2)					)

									if not return_dict:
														return (sample,)

									return FlaxUNetaDConditionOutput(sample=lowercase_					)

