


import gc
import unittest

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

from diffusers import (
    AutoencoderKL,
    DDIMScheduler,
    EulerAncestralDiscreteScheduler,
    LMSDiscreteScheduler,
    PNDMScheduler,
    StableDiffusionPanoramaPipeline,
    UNetaDConditionModel,
)
from diffusers.utils import slow, torch_device
from diffusers.utils.testing_utils import enable_full_determinism, require_torch_gpu, skip_mps

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


enable_full_determinism()



@skip_mps
class 							A   (	UpperCAmelCase__			,  UpperCAmelCase__			,  unittest.TestCase							):
					'''simple docstring'''






					A__      =			StableDiffusionPanoramaPipeline
					A__      =			TEXT_TO_IMAGE_PARAMS
					A__      =			TEXT_TO_IMAGE_BATCH_PARAMS
					A__      =			TEXT_TO_IMAGE_IMAGE_PARAMS
					A__      =			TEXT_TO_IMAGE_IMAGE_PARAMS

					def   lowerCamelCase__					(self   :						Optional[int]		)						->       str:




							"""simple docstring"""


							torch.manual_seed(0		)
							lowercase__					      =		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				, )
							lowercase__					      =		DDIMScheduler()
							torch.manual_seed(0		)
							lowercase__					      =		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				, )
							torch.manual_seed(0		)
							lowercase__					      =		CLIPTextConfig(
							    bos_token_id=0				, eos_token_id=2				, hidden_size=32				, intermediate_size=37				, layer_norm_eps=1E-0_5				, num_attention_heads=4				, num_hidden_layers=5				, pad_token_id=1				, vocab_size=1000				, )
							lowercase__					      =		CLIPTextModel(_UpperCAmelCase		)
							lowercase__					      =		CLIPTokenizer.from_pretrained("""hf-internal-testing/tiny-random-clip"""		)

							lowercase__					      =		{
							    """unet""": unet,
							    """scheduler""": scheduler,
							    """vae""": vae,
							    """text_encoder""": text_encoder,
							    """tokenizer""": tokenizer,
							    """safety_checker""": None,
							    """feature_extractor""": None,
							}
							return components

					def   lowerCamelCase__					(self   :						Dict				, _UpperCAmelCase   :						List[Any]				, _UpperCAmelCase   :						Union[str, Any]=0		)						->       Tuple:




							"""simple docstring"""


							lowercase__					      =		torch.manual_seed(_UpperCAmelCase		)
							lowercase__					      =		{
							    """prompt""": """a photo of the dolomites""",
							    """generator""": generator,
							    # Setting height and width to None to prevent OOMs on CPU.
							    """height""": None,
							    """width""": None,
							    """num_inference_steps""": 1,
							    """guidance_scale""": 6.0,
							    """output_type""": """numpy""",
							}
							return inputs

					def   lowerCamelCase__					(self   :						str		)						->       Dict:




							"""simple docstring"""


							lowercase__					      =		"""cpu"""  # ensure determinism for the device-dependent torch.Generator
							lowercase__					      =		self.get_dummy_components()
							lowercase__					      =		StableDiffusionPanoramaPipeline(**_UpperCAmelCase		)
							lowercase__					      =		sd_pipe.to(_UpperCAmelCase		)
							sd_pipe.set_progress_bar_config(disable=_UpperCAmelCase		)

							lowercase__					      =		self.get_dummy_inputs(_UpperCAmelCase		)
							lowercase__					      =		sd_pipe(**_UpperCAmelCase		).images
							lowercase__					      =		image[0, -3:, -3:, -1]
							assert image.shape == (1, 64, 64, 3)

							lowercase__					      =		np.array([0.6_186, 0.5_374, 0.4_915, 0.4_135, 0.4_114, 0.4_563, 0.5_128, 0.4_977, 0.4_757]		)

							assert np.abs(image_slice.flatten() - expected_slice		).max() < 1E-2

					def   lowerCamelCase__					(self   :						Optional[int]		)						->       Optional[Any]:




							"""simple docstring"""


							super().test_inference_batch_consistent(batch_sizes=[1, 2]		)

					def   lowerCamelCase__					(self   :						Optional[int]		)						->       Optional[Any]:




							"""simple docstring"""


							super().test_inference_batch_single_identical(batch_size=2				, expected_max_diff=3.2_5E-3		)

					def   lowerCamelCase__					(self   :						List[str]		)						->       Dict:




							"""simple docstring"""


							lowercase__					      =		"""cpu"""  # ensure determinism for the device-dependent torch.Generator
							lowercase__					      =		self.get_dummy_components()
							lowercase__					      =		StableDiffusionPanoramaPipeline(**_UpperCAmelCase		)
							lowercase__					      =		sd_pipe.to(_UpperCAmelCase		)
							sd_pipe.set_progress_bar_config(disable=_UpperCAmelCase		)

							lowercase__					      =		self.get_dummy_inputs(_UpperCAmelCase		)
							lowercase__					      =		"""french fries"""
							lowercase__					      =		sd_pipe(**_UpperCAmelCase				, negative_prompt=_UpperCAmelCase		)
							lowercase__					      =		output.images
							lowercase__					      =		image[0, -3:, -3:, -1]

							assert image.shape == (1, 64, 64, 3)

							lowercase__					      =		np.array([0.6_187, 0.5_375, 0.4_915, 0.4_136, 0.4_114, 0.4_563, 0.5_128, 0.4_976, 0.4_757]		)

							assert np.abs(image_slice.flatten() - expected_slice		).max() < 1E-2

					def   lowerCamelCase__					(self   :						Union[str, Any]		)						->       Optional[Any]:




							"""simple docstring"""


							lowercase__					      =		"""cpu"""  # ensure determinism for the device-dependent torch.Generator
							lowercase__					      =		self.get_dummy_components()
							lowercase__					      =		StableDiffusionPanoramaPipeline(**_UpperCAmelCase		)
							lowercase__					      =		sd_pipe.to(_UpperCAmelCase		)
							sd_pipe.set_progress_bar_config(disable=_UpperCAmelCase		)

							lowercase__					      =		self.get_dummy_inputs(_UpperCAmelCase		)
							lowercase__					      =		sd_pipe(**_UpperCAmelCase				, view_batch_size=2		)
							lowercase__					      =		output.images
							lowercase__					      =		image[0, -3:, -3:, -1]

							assert image.shape == (1, 64, 64, 3)

							lowercase__					      =		np.array([0.6_187, 0.5_375, 0.4_915, 0.4_136, 0.4_114, 0.4_563, 0.5_128, 0.4_976, 0.4_757]		)

							assert np.abs(image_slice.flatten() - expected_slice		).max() < 1E-2

					def   lowerCamelCase__					(self   :						List[Any]		)						->       Optional[Any]:




							"""simple docstring"""


							lowercase__					      =		"""cpu"""  # ensure determinism for the device-dependent torch.Generator
							lowercase__					      =		self.get_dummy_components()
							lowercase__					      =		EulerAncestralDiscreteScheduler(
							    beta_start=0.00_085				, beta_end=0.012				, beta_schedule="""scaled_linear"""		)
							lowercase__					      =		StableDiffusionPanoramaPipeline(**_UpperCAmelCase		)
							lowercase__					      =		sd_pipe.to(_UpperCAmelCase		)
							sd_pipe.set_progress_bar_config(disable=_UpperCAmelCase		)

							lowercase__					      =		self.get_dummy_inputs(_UpperCAmelCase		)
							lowercase__					      =		sd_pipe(**_UpperCAmelCase		).images
							lowercase__					      =		image[0, -3:, -3:, -1]

							assert image.shape == (1, 64, 64, 3)

							lowercase__					      =		np.array([0.4_024, 0.6_510, 0.4_901, 0.5_378, 0.5_813, 0.5_622, 0.4_795, 0.4_467, 0.4_952]		)

							assert np.abs(image_slice.flatten() - expected_slice		).max() < 1E-2


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




							"""simple docstring"""


							lowercase__					      =		"""cpu"""  # ensure determinism for the device-dependent torch.Generator
							lowercase__					      =		self.get_dummy_components()
							lowercase__					      =		PNDMScheduler(
							    beta_start=0.00_085				, beta_end=0.012				, beta_schedule="""scaled_linear"""				, skip_prk_steps=_UpperCAmelCase		)
							lowercase__					      =		StableDiffusionPanoramaPipeline(**_UpperCAmelCase		)
							lowercase__					      =		sd_pipe.to(_UpperCAmelCase		)
							sd_pipe.set_progress_bar_config(disable=_UpperCAmelCase		)

							lowercase__					      =		self.get_dummy_inputs(_UpperCAmelCase		)
							lowercase__					      =		sd_pipe(**_UpperCAmelCase		).images
							lowercase__					      =		image[0, -3:, -3:, -1]

							assert image.shape == (1, 64, 64, 3)

							lowercase__					      =		np.array([0.6_391, 0.6_291, 0.4_861, 0.5_134, 0.5_552, 0.4_578, 0.5_032, 0.5_023, 0.4_539]		)

							assert np.abs(image_slice.flatten() - expected_slice		).max() < 1E-2






@slow
@require_torch_gpu
class 							A   (	unittest.TestCase							):
					'''simple docstring'''






					def   lowerCamelCase__					(self   :						str		)						->       str:




							"""simple docstring"""


							super().tearDown()
							gc.collect()
							torch.cuda.empty_cache()

					def   lowerCamelCase__					(self   :						Tuple				, _UpperCAmelCase   :						Optional[int]=0		)						->       Optional[Any]:




							"""simple docstring"""


							lowercase__					      =		torch.manual_seed(_UpperCAmelCase		)
							lowercase__					      =		{
							    """prompt""": """a photo of the dolomites""",
							    """generator""": generator,
							    """num_inference_steps""": 3,
							    """guidance_scale""": 7.5,
							    """output_type""": """numpy""",
							}
							return inputs

					def   lowerCamelCase__					(self   :						Tuple		)						->       List[Any]:




							"""simple docstring"""


							lowercase__					      =		"""stabilityai/stable-diffusion-2-base"""
							lowercase__					      =		DDIMScheduler.from_pretrained(_UpperCAmelCase				, subfolder="""scheduler"""		)
							lowercase__					      =		StableDiffusionPanoramaPipeline.from_pretrained(_UpperCAmelCase				, scheduler=_UpperCAmelCase				, safety_checker=_UpperCAmelCase		)
							pipe.to(_UpperCAmelCase		)
							pipe.set_progress_bar_config(disable=_UpperCAmelCase		)
							pipe.enable_attention_slicing()

							lowercase__					      =		self.get_inputs()
							lowercase__					      =		pipe(**_UpperCAmelCase		).images
							lowercase__					      =		image[0, -3:, -3:, -1].flatten()

							assert image.shape == (1, 512, 2048, 3)

							lowercase__					      =		np.array(
							    [
							        0.36_968_392,
							        0.27_025_372,
							        0.32_446_766,
							        0.28_379_387,
							        0.36_363_274,
							        0.30_733_347,
							        0.27_100_027,
							        0.27_054_125,
							        0.25_536_096,
							    ]		)

							assert np.abs(expected_slice - image_slice		).max() < 1E-2

					def   lowerCamelCase__					(self   :						Union[str, Any]		)						->       Tuple:




							"""simple docstring"""


							lowercase__					      =		StableDiffusionPanoramaPipeline.from_pretrained(
							    """stabilityai/stable-diffusion-2-base"""				, safety_checker=_UpperCAmelCase		)
							lowercase__					      =		LMSDiscreteScheduler.from_config(pipe.scheduler.config		)
							pipe.to(_UpperCAmelCase		)
							pipe.set_progress_bar_config(disable=_UpperCAmelCase		)
							pipe.enable_attention_slicing()

							lowercase__					      =		self.get_inputs()
							lowercase__					      =		pipe(**_UpperCAmelCase		).images
							lowercase__					      =		image[0, -3:, -3:, -1].flatten()

							assert image.shape == (1, 512, 2048, 3)

							lowercase__					      =		np.array(
							    [
							        [
							            0.0,
							            0.0,
							            0.0,
							            0.0,
							            0.0,
							            0.0,
							            0.0,
							            0.0,
							            0.0,
							        ]
							    ]		)

							assert np.abs(expected_slice - image_slice		).max() < 1E-3

					def   lowerCamelCase__					(self   :						str		)						->       Any:




							"""simple docstring"""


							lowercase__					      =		0

							def callback_fn(_UpperCAmelCase   :						int				, _UpperCAmelCase   :						int				, _UpperCAmelCase   :						torch.FloatTensor		) -> None:
									lowercase__					      =		True
									nonlocal number_of_steps
									number_of_steps += 1
									if step == 1:
											lowercase__					      =		latents.detach().cpu().numpy()
											assert latents.shape == (1, 4, 64, 256)
											lowercase__					      =		latents[0, -3:, -3:, -1]

											lowercase__					      =		np.array(
											    [
											        0.18_681_869,
											        0.33_907_816,
											        0.5_361_276,
											        0.14_432_865,
											        -0.02_856_611,
											        -0.73_941_123,
											        0.23_397_987,
											        0.47_322_682,
											        -0.37_823_164,
											    ]		)
											assert np.abs(latents_slice.flatten() - expected_slice		).max() < 5E-2
									elif step == 2:
											lowercase__					      =		latents.detach().cpu().numpy()
											assert latents.shape == (1, 4, 64, 256)
											lowercase__					      =		latents[0, -3:, -3:, -1]

											lowercase__					      =		np.array(
											    [
											        0.18_539_645,
											        0.33_987_248,
											        0.5_378_559,
											        0.14_437_142,
											        -0.02_455_261,
											        -0.7_338_317,
											        0.23_990_755,
											        0.47_356_272,
											        -0.3_786_505,
											    ]		)

											assert np.abs(latents_slice.flatten() - expected_slice		).max() < 5E-2

							lowercase__					      =		False

							lowercase__					      =		"""stabilityai/stable-diffusion-2-base"""
							lowercase__					      =		DDIMScheduler.from_pretrained(_UpperCAmelCase				, subfolder="""scheduler"""		)
							lowercase__					      =		StableDiffusionPanoramaPipeline.from_pretrained(_UpperCAmelCase				, scheduler=_UpperCAmelCase				, safety_checker=_UpperCAmelCase		)
							lowercase__					      =		pipe.to(_UpperCAmelCase		)
							pipe.set_progress_bar_config(disable=_UpperCAmelCase		)
							pipe.enable_attention_slicing()

							lowercase__					      =		self.get_inputs()
							pipe(**_UpperCAmelCase				, callback=_UpperCAmelCase				, callback_steps=1		)
							assert callback_fn.has_been_called
							assert number_of_steps == 3


					def   lowerCamelCase__					(self   :						Dict		)						->       List[Any]:




							"""simple docstring"""


							torch.cuda.empty_cache()
							torch.cuda.reset_max_memory_allocated()
							torch.cuda.reset_peak_memory_stats()

							lowercase__					      =		"""stabilityai/stable-diffusion-2-base"""
							lowercase__					      =		DDIMScheduler.from_pretrained(_UpperCAmelCase				, subfolder="""scheduler"""		)
							lowercase__					      =		StableDiffusionPanoramaPipeline.from_pretrained(_UpperCAmelCase				, scheduler=_UpperCAmelCase				, safety_checker=_UpperCAmelCase		)
							lowercase__					      =		pipe.to(_UpperCAmelCase		)
							pipe.set_progress_bar_config(disable=_UpperCAmelCase		)
							pipe.enable_attention_slicing(1		)
							pipe.enable_sequential_cpu_offload()

							lowercase__					      =		self.get_inputs()
							lowercase__					      =		pipe(**_UpperCAmelCase		)

							lowercase__					      =		torch.cuda.max_memory_allocated()
							# make sure that less than 5.2 GB is allocated
							assert mem_bytes < 5.5 * 10**9





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 							A   (	unittest.TestCase							):
					'''simple docstring'''






					def   lowerCamelCase__					(self   :						Any				, _UpperCAmelCase   :						List[str]				, _UpperCAmelCase   :						Union[str, Any]		)						->       int:




							"""simple docstring"""


							return f'''gaussian_noise_s={seed}_shape={"_".join([str(_UpperCAmelCase		) for s in shape]		)}.npy'''

					def   lowerCamelCase__					(self   :						Any		)						->       int:




							"""simple docstring"""


							super().tearDown()
							gc.collect()

					def   lowerCamelCase__					(self   :						List[Any]				, _UpperCAmelCase   :						Any=0				, _UpperCAmelCase   :						Optional[Any]=(4, 4, 64, 64)				, _UpperCAmelCase   :						Any=False		)						->       List[Any]:




							"""simple docstring"""


							lowercase__					      =		jnp.bfloataa if fpaa else jnp.floataa
							lowercase__					      =		jnp.array(load_hf_numpy(self.get_file_format(_UpperCAmelCase				, _UpperCAmelCase		)		)				, dtype=_UpperCAmelCase		)
							return image

					def   lowerCamelCase__					(self   :						Optional[Any]				, _UpperCAmelCase   :						str=False				, _UpperCAmelCase   :						str="CompVis/stable-diffusion-v1-4"		)						->       Dict:




							"""simple docstring"""


							lowercase__					      =		jnp.bfloataa if fpaa else jnp.floataa
							lowercase__					      =		"""bf16""" if fpaa else None

							lowercase__      ,     lowercase__					      =		FlaxUNetaDConditionModel.from_pretrained(
							    _UpperCAmelCase				, subfolder="""unet"""				, dtype=_UpperCAmelCase				, revision=_UpperCAmelCase		)
							return model, params

					def   lowerCamelCase__					(self   :						List[Any]				, _UpperCAmelCase   :						int=0				, _UpperCAmelCase   :						Optional[int]=(4, 77, 768)				, _UpperCAmelCase   :						Any=False		)						->       Optional[Any]:




							"""simple docstring"""


							lowercase__					      =		jnp.bfloataa if fpaa else jnp.floataa
							lowercase__					      =		jnp.array(load_hf_numpy(self.get_file_format(_UpperCAmelCase				, _UpperCAmelCase		)		)				, dtype=_UpperCAmelCase		)
							return hidden_states

					@parameterized.expand(
					    [
					        # fmt: off
					        [83, 4, [-0.2_323, -0.1_304, 0.0_813, -0.3_093, -0.0_919, -0.1_571, -0.1_125, -0.5_806]],
					        [17, 0.55, [-0.0_831, -0.2_443, 0.0_901, -0.0_919, 0.3_396, 0.0_103, -0.3_743, 0.0_701]],
					        [8, 0.89, [-0.4_863, 0.0_859, 0.0_875, -0.1_658, 0.9_199, -0.0_114, 0.4_839, 0.4_639]],
					        [3, 1000, [-0.5_649, 0.2_402, -0.5_518, 0.1_248, 1.1_328, -0.2_443, -0.0_325, -1.0_078]],
					        # fmt: on
					    ]		)
					def   lowerCamelCase__					(self   :						int				, _UpperCAmelCase   :						str				, _UpperCAmelCase   :						Union[str, Any]				, _UpperCAmelCase   :						int		)						->       str:




							"""simple docstring"""


							lowercase__      ,     lowercase__					      =		self.get_unet_model(model_id="""CompVis/stable-diffusion-v1-4"""				, fpaa=_UpperCAmelCase		)
							lowercase__					      =		self.get_latents(_UpperCAmelCase				, fpaa=_UpperCAmelCase		)
							lowercase__					      =		self.get_encoder_hidden_states(_UpperCAmelCase				, fpaa=_UpperCAmelCase		)

							lowercase__					      =		model.apply(
							    {"""params""": params}				, _UpperCAmelCase				, jnp.array(_UpperCAmelCase				, dtype=jnp.intaa		)				, encoder_hidden_states=_UpperCAmelCase				, ).sample

							assert sample.shape == latents.shape

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

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


					@parameterized.expand(
					    [
					        # fmt: off
					        [83, 4, [0.1_514, 0.0_807, 0.1_624, 0.1_016, -0.1_896, 0.0_263, 0.0_677, 0.2_310]],
					        [17, 0.55, [0.1_164, -0.0_216, 0.0_170, 0.1_589, -0.3_120, 0.1_005, -0.0_581, -0.1_458]],
					        [8, 0.89, [-0.1_758, -0.0_169, 0.1_004, -0.1_411, 0.1_312, 0.1_103, -0.1_996, 0.2_139]],
					        [3, 1000, [0.1_214, 0.0_352, -0.0_731, -0.1_562, -0.0_994, -0.0_906, -0.2_340, -0.0_539]],
					        # fmt: on
					    ]		)
					def   lowerCamelCase__					(self   :						Optional[Any]				, _UpperCAmelCase   :						Tuple				, _UpperCAmelCase   :						int				, _UpperCAmelCase   :						List[str]		)						->       str:




							"""simple docstring"""


							lowercase__      ,     lowercase__					      =		self.get_unet_model(model_id="""stabilityai/stable-diffusion-2"""				, fpaa=_UpperCAmelCase		)
							lowercase__					      =		self.get_latents(_UpperCAmelCase				, shape=(4, 4, 96, 96)				, fpaa=_UpperCAmelCase		)
							lowercase__					      =		self.get_encoder_hidden_states(_UpperCAmelCase				, shape=(4, 77, 1024)				, fpaa=_UpperCAmelCase		)

							lowercase__					      =		model.apply(
							    {"""params""": params}				, _UpperCAmelCase				, jnp.array(_UpperCAmelCase				, dtype=jnp.intaa		)				, encoder_hidden_states=_UpperCAmelCase				, ).sample

							assert sample.shape == latents.shape

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

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

