

import gc
import unittest

from parameterized import parameterized

from diffusers import FlaxUNetaDConditionModel
from diffusers.utils import is_flax_available
from diffusers.utils.testing_utils import load_hf_numpy, require_flax, slow


if is_flax_available():
		import jax
		import jax.numpy as jnp





@slow
@require_flax
class     snake_case_			(				unittest.TestCase							):


							"""simple docstring"""





							def        A__  ( self					,						_A					,						_A							):
												return F"""gaussian_noise_s={seed}_shape={'_'.join([str(_A							) for s in shape]							)}.npy"""




							def        A__  ( self							):
												# clean up the VRAM after each test
												super().tearDown()
												gc.collect()




							def        A__  ( self					,						_A=0					,						_A=(4, 4, 6_4, 6_4)					,						_A=False							):
												__lowerCAmelCase					=							jnp.bfloataa if fpaa else jnp.floataa
												__lowerCAmelCase					=							jnp.array(load_hf_numpy(self.get_file_format(_A					,						_A							)							)					,						dtype=_A							)
												return image




							def        A__  ( self					,						_A=False					,						_A="CompVis/stable-diffusion-v1-4"							):
												__lowerCAmelCase					=							jnp.bfloataa if fpaa else jnp.floataa
												__lowerCAmelCase					=							'bf16' if fpaa else None

												__lowerCAmelCase,	__lowerCAmelCase					=							FlaxUNetaDConditionModel.from_pretrained(
												    _A					,						subfolder='unet'					,						dtype=_A					,						revision=_A							)
												return model, params




							def        A__  ( self					,						_A=0					,						_A=(4, 7_7, 7_6_8)					,						_A=False							):
												__lowerCAmelCase					=							jnp.bfloataa if fpaa else jnp.floataa
												__lowerCAmelCase					=							jnp.array(load_hf_numpy(self.get_file_format(_A					,						_A							)							)					,						dtype=_A							)
												return hidden_states




							@parameterized.expand(
							    [
							        # fmt: off
							        [8_3, 4, [-0.2323, -0.1304, 0.0813, -0.3093, -0.0919, -0.1571, -0.1125, -0.5806]],
							        [1_7, 0.55, [-0.0831, -0.2443, 0.0901, -0.0919, 0.3396, 0.0103, -0.3743, 0.0701]],
							        [8, 0.89, [-0.4863, 0.0859, 0.0875, -0.1658, 0.9199, -0.0114, 0.4839, 0.4639]],
							        [3, 1_0_0_0, [-0.5649, 0.2402, -0.5518, 0.1248, 1.1328, -0.2443, -0.0325, -1.0078]],
							        # fmt: on
							    ]							)
							def        A__  ( self					,						_A					,						_A					,						_A							):
												__lowerCAmelCase,	__lowerCAmelCase					=							self.get_unet_model(model_id='CompVis/stable-diffusion-v1-4'					,						fpaa=_A							)
												__lowerCAmelCase					=							self.get_latents(_A					,						fpaa=_A							)
												__lowerCAmelCase					=							self.get_encoder_hidden_states(_A					,						fpaa=_A							)

												__lowerCAmelCase					=							model.apply(
												    {'params': params}					,						_A					,						jnp.array(_A					,						dtype=jnp.intaa							)					,						encoder_hidden_states=_A					,						).sample

												assert sample.shape == latents.shape

												__lowerCAmelCase					=							jnp.asarray(jax.device_get((sample[-1, -2:, -2:, :2].flatten())							)					,						dtype=jnp.floataa							)
												__lowerCAmelCase					=							jnp.array(_A					,						dtype=jnp.floataa							)

												# Found torch (float16) and flax (bfloat16) outputs to be within this tolerance, in the same hardware
												assert jnp.allclose(_A					,						_A					,						atol=1e-2							)




							@parameterized.expand(
							    [
							        # fmt: off
							        [8_3, 4, [0.1514, 0.0807, 0.1624, 0.1016, -0.1896, 0.0263, 0.0677, 0.2310]],
							        [1_7, 0.55, [0.1164, -0.0216, 0.0170, 0.1589, -0.3120, 0.1005, -0.0581, -0.1458]],
							        [8, 0.89, [-0.1758, -0.0169, 0.1004, -0.1411, 0.1312, 0.1103, -0.1996, 0.2139]],
							        [3, 1_0_0_0, [0.1214, 0.0352, -0.0731, -0.1562, -0.0994, -0.0906, -0.2340, -0.0539]],
							        # fmt: on
							    ]							)
							def        A__  ( self					,						_A					,						_A					,						_A							):
												__lowerCAmelCase,	__lowerCAmelCase					=							self.get_unet_model(model_id='stabilityai/stable-diffusion-2'					,						fpaa=_A							)
												__lowerCAmelCase					=							self.get_latents(_A					,						shape=(4, 4, 9_6, 9_6)					,						fpaa=_A							)
												__lowerCAmelCase					=							self.get_encoder_hidden_states(_A					,						shape=(4, 7_7, 1_0_2_4)					,						fpaa=_A							)

												__lowerCAmelCase					=							model.apply(
												    {'params': params}					,						_A					,						jnp.array(_A					,						dtype=jnp.intaa							)					,						encoder_hidden_states=_A					,						).sample

												assert sample.shape == latents.shape

												__lowerCAmelCase					=							jnp.asarray(jax.device_get((sample[-1, -2:, -2:, :2].flatten())							)					,						dtype=jnp.floataa							)
												__lowerCAmelCase					=							jnp.array(_A					,						dtype=jnp.floataa							)

												# Found torch (float16) and flax (bfloat16) outputs to be within this tolerance, on the same hardware
												assert jnp.allclose(_A					,						_A					,						atol=1e-2							)









from dataclasses import dataclass
from typing import Dict, Optional, Tuple, Union

import torch
import torch.nn as nn

from ..configuration_utils import ConfigMixin, register_to_config
from ..utils import BaseOutput, apply_forward_hook
from .attention_processor import AttentionProcessor, AttnProcessor
from .modeling_utils import ModelMixin
from .vae import Decoder, DecoderOutput, DiagonalGaussianDistribution, Encoder





@dataclass
class     snake_case_			(				_a							):


							"""simple docstring"""





							__UpperCAmelCase       =42





class     snake_case_			(				_a			, _a							):


							"""simple docstring"""





							__UpperCAmelCase       =True




							@register_to_config
							def __init__( self					,						_A = 3					,						_A = 3					,						_A = ("DownEncoderBlock2D",)					,						_A = ("UpDecoderBlock2D",)					,						_A = (6_4,)					,						_A = 1					,						_A = "silu"					,						_A = 4					,						_A = 3_2					,						_A = 3_2					,						_A = 0.1_8215					,						):
												super().__init__()

												# pass init params to Encoder
												__lowerCAmelCase					=							Encoder(
												    in_channels=_A					,						out_channels=_A					,						down_block_types=_A					,						block_out_channels=_A					,						layers_per_block=_A					,						act_fn=_A					,						norm_num_groups=_A					,						double_z=_A					,						)

												# pass init params to Decoder
												__lowerCAmelCase					=							Decoder(
												    in_channels=_A					,						out_channels=_A					,						up_block_types=_A					,						block_out_channels=_A					,						layers_per_block=_A					,						norm_num_groups=_A					,						act_fn=_A					,						)

												__lowerCAmelCase					=							nn.Convad(2 * latent_channels					,						2 * latent_channels					,						1							)
												__lowerCAmelCase					=							nn.Convad(_A					,						_A					,						1							)

												__lowerCAmelCase					=							False
												__lowerCAmelCase					=							False

												# only relevant if vae tiling is enabled
												__lowerCAmelCase					=							self.config.sample_size
												__lowerCAmelCase					=							(
												    self.config.sample_size[0]
												    if isinstance(self.config.sample_size					,						(list, tuple)							)
												    else self.config.sample_size
												)
												__lowerCAmelCase					=							int(sample_size / (2 ** (len(self.config.block_out_channels							) - 1))							)
												__lowerCAmelCase					=							0.25




							def        A__  ( self					,						_A					,						_A=False							):
												if isinstance(_A					,						(Encoder, Decoder)							):
																	__lowerCAmelCase					=							value




							def        A__  ( self					,						_A = True							):
												__lowerCAmelCase					=							use_tiling




							def        A__  ( self							):
												self.enable_tiling(_A							)




							def        A__  ( self							):
												__lowerCAmelCase					=							True




							def        A__  ( self							):
												__lowerCAmelCase					=							False




							@property
							# Copied from diffusers.models.unet_2d_condition.UNet2DConditionModel.attn_processors
							def        A__  ( self							):
												__lowerCAmelCase					=							{}

												def fn_recursive_add_processors(_A					,						_A					,						_A							):
																	if hasattr(_A					,						'set_processor'							):
																						__lowerCAmelCase					=							module.processor

																	for sub_name, child in module.named_children():
																						fn_recursive_add_processors(F"""{name}.{sub_name}"""					,						_A					,						_A							)

																	return processors

												for name, module in self.named_children():
																	fn_recursive_add_processors(_A					,						_A					,						_A							)

												return processors




							def        A__  ( self					,						_A							):
												__lowerCAmelCase					=							len(self.attn_processors.keys()							)

												if isinstance(_A					,						_A							) and len(_A							) != count:
																	raise ValueError(
																	    F"""A dict of processors was passed, but the number of processors {len(_A							)} does not match the"""
																	    F""" number of attention layers: {count}. Please make sure to pass {count} processor classes."""							)

												def fn_recursive_attn_processor(_A					,						_A					,						_A							):
																	if hasattr(_A					,						'set_processor'							):
																						if not isinstance(_A					,						_A							):
																											module.set_processor(_A							)
																						else:
																											module.set_processor(processor.pop(F"""{name}.processor"""							)							)

																	for sub_name, child in module.named_children():
																						fn_recursive_attn_processor(F"""{name}.{sub_name}"""					,						_A					,						_A							)

												for name, module in self.named_children():
																	fn_recursive_attn_processor(_A					,						_A					,						_A							)




							def        A__  ( self							):
												self.set_attn_processor(AttnProcessor()							)




							@apply_forward_hook
							def        A__  ( self					,						_A					,						_A = True							):
												if self.use_tiling and (x.shape[-1] > self.tile_sample_min_size or x.shape[-2] > self.tile_sample_min_size):
																	return self.tiled_encode(_A					,						return_dict=_A							)

												if self.use_slicing and x.shape[0] > 1:
																	__lowerCAmelCase					=							[self.encoder(_A							) for x_slice in x.split(1							)]
																	__lowerCAmelCase					=							torch.cat(_A							)
												else:
																	__lowerCAmelCase					=							self.encoder(_A							)

												__lowerCAmelCase					=							self.quant_conv(_A							)
												__lowerCAmelCase					=							DiagonalGaussianDistribution(_A							)

												if not return_dict:
																	return (posterior,)

												return AutoencoderKLOutput(latent_dist=_A							)




							def        A__  ( self					,						_A					,						_A = True							):
												if self.use_tiling and (z.shape[-1] > self.tile_latent_min_size or z.shape[-2] > self.tile_latent_min_size):
																	return self.tiled_decode(_A					,						return_dict=_A							)

												__lowerCAmelCase					=							self.post_quant_conv(_A							)
												__lowerCAmelCase					=							self.decoder(_A							)

												if not return_dict:
																	return (dec,)

												return DecoderOutput(sample=_A							)




							@apply_forward_hook
							def        A__  ( self					,						_A					,						_A = True							):
												if self.use_slicing and z.shape[0] > 1:
																	__lowerCAmelCase					=							[self._decode(_A							).sample for z_slice in z.split(1							)]
																	__lowerCAmelCase					=							torch.cat(_A							)
												else:
																	__lowerCAmelCase					=							self._decode(_A							).sample

												if not return_dict:
																	return (decoded,)

												return DecoderOutput(sample=_A							)




							def        A__  ( self					,						_A					,						_A					,						_A							):
												__lowerCAmelCase					=							min(a.shape[2]					,						b.shape[2]					,						_A							)
												for y in range(_A							):
																	__lowerCAmelCase					=							a[:, :, -blend_extent + y, :] * (1 - y / blend_extent) + b[:, :, y, :] * (y / blend_extent)
												return b




							def        A__  ( self					,						_A					,						_A					,						_A							):
												__lowerCAmelCase					=							min(a.shape[3]					,						b.shape[3]					,						_A							)
												for x in range(_A							):
																	__lowerCAmelCase					=							a[:, :, :, -blend_extent + x] * (1 - x / blend_extent) + b[:, :, :, x] * (x / blend_extent)
												return b




							def        A__  ( self					,						_A					,						_A = True							):
												__lowerCAmelCase					=							int(self.tile_sample_min_size * (1 - self.tile_overlap_factor)							)
												__lowerCAmelCase					=							int(self.tile_latent_min_size * self.tile_overlap_factor							)
												__lowerCAmelCase					=							self.tile_latent_min_size - blend_extent

												# Split the image into 512x512 tiles and encode them separately.
												__lowerCAmelCase					=							[]
												for i in range(0					,						x.shape[2]					,						_A							):
																	__lowerCAmelCase					=							[]
																	for j in range(0					,						x.shape[3]					,						_A							):
																						__lowerCAmelCase					=							x[:, :, i : i + self.tile_sample_min_size, j : j + self.tile_sample_min_size]
																						__lowerCAmelCase					=							self.encoder(_A							)
																						__lowerCAmelCase					=							self.quant_conv(_A							)
																						row.append(_A							)
																	rows.append(_A							)
												__lowerCAmelCase					=							[]
												for i, row in enumerate(_A							):
																	__lowerCAmelCase					=							[]
																	for j, tile in enumerate(_A							):
																						# blend the above tile and the left tile
																						# to the current tile and add the current tile to the result row
																						if i > 0:
																											__lowerCAmelCase					=							self.blend_v(rows[i - 1][j]					,						_A					,						_A							)
																						if j > 0:
																											__lowerCAmelCase					=							self.blend_h(row[j - 1]					,						_A					,						_A							)
																						result_row.append(tile[:, :, :row_limit, :row_limit]							)
																	result_rows.append(torch.cat(_A					,						dim=3							)							)

												__lowerCAmelCase					=							torch.cat(_A					,						dim=2							)
												__lowerCAmelCase					=							DiagonalGaussianDistribution(_A							)

												if not return_dict:
																	return (posterior,)

												return AutoencoderKLOutput(latent_dist=_A							)




							def        A__  ( self					,						_A					,						_A = True							):
												__lowerCAmelCase					=							int(self.tile_latent_min_size * (1 - self.tile_overlap_factor)							)
												__lowerCAmelCase					=							int(self.tile_sample_min_size * self.tile_overlap_factor							)
												__lowerCAmelCase					=							self.tile_sample_min_size - blend_extent

												# Split z into overlapping 64x64 tiles and decode them separately.
												# The tiles have an overlap to avoid seams between tiles.
												__lowerCAmelCase					=							[]
												for i in range(0					,						z.shape[2]					,						_A							):
																	__lowerCAmelCase					=							[]
																	for j in range(0					,						z.shape[3]					,						_A							):
																						__lowerCAmelCase					=							z[:, :, i : i + self.tile_latent_min_size, j : j + self.tile_latent_min_size]
																						__lowerCAmelCase					=							self.post_quant_conv(_A							)
																						__lowerCAmelCase					=							self.decoder(_A							)
																						row.append(_A							)
																	rows.append(_A							)
												__lowerCAmelCase					=							[]
												for i, row in enumerate(_A							):
																	__lowerCAmelCase					=							[]
																	for j, tile in enumerate(_A							):
																						# blend the above tile and the left tile
																						# to the current tile and add the current tile to the result row
																						if i > 0:
																											__lowerCAmelCase					=							self.blend_v(rows[i - 1][j]					,						_A					,						_A							)
																						if j > 0:
																											__lowerCAmelCase					=							self.blend_h(row[j - 1]					,						_A					,						_A							)
																						result_row.append(tile[:, :, :row_limit, :row_limit]							)
																	result_rows.append(torch.cat(_A					,						dim=3							)							)

												__lowerCAmelCase					=							torch.cat(_A					,						dim=2							)
												if not return_dict:
																	return (dec,)

												return DecoderOutput(sample=_A							)




							def        A__  ( self					,						_A					,						_A = False					,						_A = True					,						_A = None					,						):
												__lowerCAmelCase					=							sample
												__lowerCAmelCase					=							self.encode(_A							).latent_dist
												if sample_posterior:
																	__lowerCAmelCase					=							posterior.sample(generator=_A							)
												else:
																	__lowerCAmelCase					=							posterior.mode()
												__lowerCAmelCase					=							self.decode(_A							).sample

												if not return_dict:
																	return (dec,)

												return DecoderOutput(sample=_A							)






