


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    _UpperCamelCase     (       __snake_case					):

				"""simple docstring"""







				lowerCAmelCase					=			42






@flax_register_to_config
class    _UpperCamelCase     (       nn.Module     ,		__snake_case     ,		__snake_case					):

				"""simple docstring"""







				lowerCAmelCase					=			3_2
				lowerCAmelCase					=			4
				lowerCAmelCase					=			4
				lowerCAmelCase					=			(
				    "CrossAttnDownBlock2D",
				    "CrossAttnDownBlock2D",
				    "CrossAttnDownBlock2D",
				    "DownBlock2D",
				)
				lowerCAmelCase					=			("UpBlock2D", "CrossAttnUpBlock2D", "CrossAttnUpBlock2D", "CrossAttnUpBlock2D")
				lowerCAmelCase					=			False
				lowerCAmelCase					=			(3_2_0, 6_4_0, 1_2_8_0, 1_2_8_0)
				lowerCAmelCase					=			2
				lowerCAmelCase					=			8
				lowerCAmelCase					=			None
				lowerCAmelCase					=			1_2_8_0
				lowerCAmelCase					=			0.0
				lowerCAmelCase					=			False
				lowerCAmelCase					=			jnp.floataa
				lowerCAmelCase					=			True
				lowerCAmelCase					=			0
				lowerCAmelCase					=			False

				def   _UpperCAmelCase       (    self					, a__ )						->					FrozenDict:
					# init input tensors
					A             =					(1, self.in_channels, self.sample_size, self.sample_size)
					A             =					jnp.zeros(a__					, dtype=jnp.floataa )
					A             =					jnp.ones((1,)					, dtype=jnp.intaa )
					A             =					jnp.zeros((1, 1, self.cross_attention_dim)					, dtype=jnp.floataa )

					A       ,      A             =					jax.random.split(a__ )
					A             =					{"""params""": params_rng, """dropout""": dropout_rng}

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

				def   _UpperCAmelCase       (    self )						->					str:
					A             =					self.block_out_channels
					A             =					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.
					A             =					self.num_attention_heads or self.attention_head_dim

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

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

					A             =					self.only_cross_attention
					if isinstance(a__					, a__ ):
						A             =					(only_cross_attention,) * len(self.down_block_types )

					if isinstance(a__					, a__ ):
						A             =					(num_attention_heads,) * len(self.down_block_types )

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

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

						down_blocks.append(a__ )
					A             =					down_blocks

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

						A             =					i == len(a__ ) - 1

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

						up_blocks.append(a__ )
						A             =					output_channel
					A             =					up_blocks

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

				def __call__(    self					, a__					, a__					, a__					, a__=None					, a__=None					, a__ = True					, a__ = False					, )						->					Union[FlaxUNetaDConditionOutput, Tuple]:
					# 1. time
					if not isinstance(a__					, jnp.ndarray ):
						A             =					jnp.array([timesteps]					, dtype=jnp.intaa )
					elif isinstance(a__					, jnp.ndarray ) and len(timesteps.shape ) == 0:
						A             =					timesteps.astype(dtype=jnp.floataa )
						A             =					jnp.expand_dims(a__					, 0 )

					A             =					self.time_proj(a__ )
					A             =					self.time_embedding(a__ )

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

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

					if down_block_additional_residuals is not None:
						A             =					()

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

						A             =					new_down_block_res_samples

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

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

					if not return_dict:
						return (sample,)

					return FlaxUNetaDConditionOutput(sample=a__ )







import gc
import unittest

import numpy as np
import torch
from transformers import CLIPTextConfig, CLIPTextModel, CLIPTokenizer

from diffusers import (
    AutoencoderKL,
    DDIMScheduler,
    StableDiffusionAttendAndExcitePipeline,
    UNetaDConditionModel,
)
from diffusers.utils import load_numpy, skip_mps, slow
from diffusers.utils.testing_utils import require_torch_gpu

from ..pipeline_params import TEXT_TO_IMAGE_BATCH_PARAMS, TEXT_TO_IMAGE_IMAGE_PARAMS, TEXT_TO_IMAGE_PARAMS
from ..test_pipelines_common import PipelineKarrasSchedulerTesterMixin, PipelineLatentTesterMixin, PipelineTesterMixin


_lowercase							:   Optional[int]						=						False






@skip_mps
class    _UpperCamelCase     (       __snake_case     ,		__snake_case     ,		__snake_case     ,		unittest.TestCase					):

				"""simple docstring"""







				lowerCAmelCase					=			StableDiffusionAttendAndExcitePipeline
				lowerCAmelCase					=			False
				lowerCAmelCase					=			TEXT_TO_IMAGE_PARAMS
				lowerCAmelCase					=			TEXT_TO_IMAGE_BATCH_PARAMS.union({'token_indices'}					)
				lowerCAmelCase					=			TEXT_TO_IMAGE_IMAGE_PARAMS
				lowerCAmelCase					=			TEXT_TO_IMAGE_IMAGE_PARAMS

				@classmethod
				def   _UpperCAmelCase       (    cls )						->					List[Any]:
					super().setUpClass()
					torch.use_deterministic_algorithms(a__ )

				@classmethod
				def   _UpperCAmelCase       (    cls )						->					Tuple:
					super().tearDownClass()
					torch.use_deterministic_algorithms(a__ )

				def   _UpperCAmelCase       (    self )						->					Dict:
					torch.manual_seed(0 )
					A             =					UNetaDConditionModel(
					    block_out_channels=(32, 64)					, layers_per_block=1					, sample_size=32					, in_channels=4					, out_channels=4					, down_block_types=("""DownBlock2D""", """CrossAttnDownBlock2D""")					, up_block_types=("""CrossAttnUpBlock2D""", """UpBlock2D""")					, cross_attention_dim=32					, attention_head_dim=(2, 4)					, use_linear_projection=a__					, )
					A             =					DDIMScheduler(
					    beta_start=0.0_00_85					, beta_end=0.0_12					, beta_schedule="""scaled_linear"""					, clip_sample=a__					, set_alpha_to_one=a__					, )
					torch.manual_seed(0 )
					A             =					AutoencoderKL(
					    block_out_channels=[32, 64]					, in_channels=3					, out_channels=3					, down_block_types=["""DownEncoderBlock2D""", """DownEncoderBlock2D"""]					, up_block_types=["""UpDecoderBlock2D""", """UpDecoderBlock2D"""]					, latent_channels=4					, sample_size=128					, )
					torch.manual_seed(0 )
					A             =					CLIPTextConfig(
					    bos_token_id=0					, eos_token_id=2					, hidden_size=32					, intermediate_size=37					, layer_norm_eps=1e-05					, num_attention_heads=4					, num_hidden_layers=5					, pad_token_id=1					, vocab_size=1000					, hidden_act="""gelu"""					, projection_dim=512					, )
					A             =					CLIPTextModel(a__ )
					A             =					CLIPTokenizer.from_pretrained("""hf-internal-testing/tiny-random-clip""" )

					A             =					{
					    """unet""": unet,
					    """scheduler""": scheduler,
					    """vae""": vae,
					    """text_encoder""": text_encoder,
					    """tokenizer""": tokenizer,
					    """safety_checker""": None,
					    """feature_extractor""": None,
					}

					return components

				def   _UpperCAmelCase       (    self					, a__					, a__=0 )						->					Optional[Any]:
					if str(a__ ).startswith("""mps""" ):
						A             =					torch.manual_seed(a__ )
					else:
						A             =					torch.Generator(device=a__ ).manual_seed(a__ )
					A             =					A             =					{
					    """prompt""": """a cat and a frog""",
					    """token_indices""": [2, 5],
					    """generator""": generator,
					    """num_inference_steps""": 1,
					    """guidance_scale""": 6.0,
					    """output_type""": """numpy""",
					    """max_iter_to_alter""": 2,
					    """thresholds""": {0: 0.7},
					}
					return inputs

				def   _UpperCAmelCase       (    self )						->					Union[str, Any]:
					A             =					"""cpu"""

					A             =					self.get_dummy_components()
					A             =					self.pipeline_class(**a__ )
					pipe.to(a__ )
					pipe.set_progress_bar_config(disable=a__ )

					A             =					self.get_dummy_inputs(a__ )
					A             =					pipe(**a__ ).images
					A             =					image[0, -3:, -3:, -1]

					self.assertEqual(image.shape					, (1, 64, 64, 3) )
					A             =					np.array(
					    [0.63_90_53_64, 0.62_89_73_07, 0.48_59_90_17, 0.5_13_36_24, 0.5_55_00_48, 0.45_76_95_16, 0.50_32_69_73, 0.5_02_31_39, 0.45_38_44_96] )
					A             =					np.abs(image_slice.flatten() - expected_slice ).max()
					self.assertLessEqual(a__					, 1e-3 )

				def   _UpperCAmelCase       (    self )						->					List[Any]:
					super().test_cpu_offload_forward_pass(expected_max_diff=5e-4 )

				def   _UpperCAmelCase       (    self )						->					Dict:
					# NOTE: Larger batch sizes cause this test to timeout, only test on smaller batches
					self._test_inference_batch_consistent(batch_sizes=[1, 2] )

				def   _UpperCAmelCase       (    self )						->					Any:
					self._test_inference_batch_single_identical(batch_size=2					, expected_max_diff=7e-4 )

				def   _UpperCAmelCase       (    self )						->					Optional[Any]:
					super().test_dict_tuple_outputs_equivalent(expected_max_difference=3e-3 )

				def   _UpperCAmelCase       (    self )						->					str:
					super().test_pt_np_pil_outputs_equivalent(expected_max_diff=5e-4 )

				def   _UpperCAmelCase       (    self )						->					int:
					super().test_save_load_local(expected_max_difference=5e-4 )

				def   _UpperCAmelCase       (    self )						->					Optional[Any]:
					super().test_save_load_optional_components(expected_max_difference=4e-4 )






@require_torch_gpu
@slow
class    _UpperCamelCase     (       unittest.TestCase					):

				"""simple docstring"""







				@classmethod
				def   _UpperCAmelCase       (    cls )						->					Tuple:
					super().setUpClass()
					torch.use_deterministic_algorithms(a__ )

				@classmethod
				def   _UpperCAmelCase       (    cls )						->					Dict:
					super().tearDownClass()
					torch.use_deterministic_algorithms(a__ )

				def   _UpperCAmelCase       (    self )						->					Optional[Any]:
					super().tearDown()
					gc.collect()
					torch.cuda.empty_cache()

				def   _UpperCAmelCase       (    self )						->					int:
					A             =					torch.manual_seed(51 )

					A             =					StableDiffusionAttendAndExcitePipeline.from_pretrained(
					    """CompVis/stable-diffusion-v1-4"""					, safety_checker=a__					, torch_dtype=torch.floataa )
					pipe.to("""cuda""" )

					A             =					"""a painting of an elephant with glasses"""
					A             =					[5, 7]

					A             =					pipe(
					    prompt=a__					, token_indices=a__					, guidance_scale=7.5					, generator=a__					, num_inference_steps=5					, max_iter_to_alter=5					, output_type="""numpy"""					, ).images[0]

					A             =					load_numpy(
					    """https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/attend-and-excite/elephant_glasses.npy""" )
					assert np.abs((expected_image - image).max() ) < 5e-1



