


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    (       __snake_case ,  __snake_case ,  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    :						List[Any]		)						->   int:




							"""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(A_		)
							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    :						List[str]      ,					_UpperCAmelCase    :						Tuple      ,					_UpperCAmelCase    :						Optional[Any]=0		)						->   int:




							"""simple docstring"""


							lowercase__     			=		torch.manual_seed(A_		)
							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		)						->   Optional[Any]:




							"""simple docstring"""


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

							lowercase__     			=		self.get_dummy_inputs(A_		)
							lowercase__     			=		sd_pipe(**A_		).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    :						List[str]		)						->   List[str]:




							"""simple docstring"""


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

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




							"""simple docstring"""


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

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




							"""simple docstring"""


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

							lowercase__     			=		self.get_dummy_inputs(A_		)
							lowercase__     			=		"french fries"
							lowercase__     			=		sd_pipe(**A_      ,					negative_prompt=A_		)
							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(**A_		)
							lowercase__     			=		sd_pipe.to(A_		)
							sd_pipe.set_progress_bar_config(disable=A_		)

							lowercase__     			=		self.get_dummy_inputs(A_		)
							lowercase__     			=		sd_pipe(**A_      ,					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    :						Any		)						->   Optional[int]:




							"""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(**A_		)
							lowercase__     			=		sd_pipe.to(A_		)
							sd_pipe.set_progress_bar_config(disable=A_		)

							lowercase__     			=		self.get_dummy_inputs(A_		)
							lowercase__     			=		sd_pipe(**A_		).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    :						Optional[int]		)						->   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=A_		)
							lowercase__     			=		StableDiffusionPanoramaPipeline(**A_		)
							lowercase__     			=		sd_pipe.to(A_		)
							sd_pipe.set_progress_bar_config(disable=A_		)

							lowercase__     			=		self.get_dummy_inputs(A_		)
							lowercase__     			=		sd_pipe(**A_		).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    :						Optional[int]		)						->   Any:




							"""simple docstring"""


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

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




							"""simple docstring"""


							lowercase__     			=		torch.manual_seed(A_		)
							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    :						Any		)						->   Optional[int]:




							"""simple docstring"""


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

							lowercase__     			=		self.get_inputs()
							lowercase__     			=		pipe(**A_		).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    :						Any		)						->   List[str]:




							"""simple docstring"""


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

							lowercase__     			=		self.get_inputs()
							lowercase__     			=		pipe(**A_		).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    :						Optional[int]		)						->   int:




							"""simple docstring"""


							lowercase__     			=		0

							def callback_fn(_UpperCAmelCase    :						Optional[int]      ,					_UpperCAmelCase    :						List[Any]      ,					_UpperCAmelCase    :						List[Any]		) -> 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(A_      ,					subfolder="""scheduler"""		)
							lowercase__     			=		StableDiffusionPanoramaPipeline.from_pretrained(A_      ,					scheduler=A_      ,					safety_checker=A_		)
							lowercase__     			=		pipe.to(A_		)
							pipe.set_progress_bar_config(disable=A_		)
							pipe.enable_attention_slicing()

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


					def   lowerCamelCase__     (self    :						int		)						->   Tuple:




							"""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(A_      ,					subfolder="""scheduler"""		)
							lowercase__     			=		StableDiffusionPanoramaPipeline.from_pretrained(A_      ,					scheduler=A_      ,					safety_checker=A_		)
							lowercase__     			=		pipe.to(A_		)
							pipe.set_progress_bar_config(disable=A_		)
							pipe.enable_attention_slicing(1		)
							pipe.enable_sequential_cpu_offload()

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

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








def 		A_				(     _lowerCAmelCase     )						-> str:
 UpperCamelCase	:					Optional[int]       	=   int(_lowerCAmelCase     )
 if decimal in (0, 1):  # Exit cases for the recursion
  return str(_lowerCAmelCase     )
 UpperCamelCase	,    UpperCamelCase	:					Dict       	=   divmod(_lowerCAmelCase							,   2     )
 return binary_recursive(_lowerCAmelCase     ) + str(_lowerCAmelCase     )



def 		A_				(     _lowerCAmelCase     )						-> str:
 UpperCamelCase	:					Tuple       	=   str(_lowerCAmelCase     ).strip()
 if not number:
  raise ValueError("No input value was provided"     )
 UpperCamelCase	:					Optional[int]       	=   "-" if number.startswith("-"     ) else ""
 UpperCamelCase	:					Any       	=   number.lstrip("-"     )
 if not number.isnumeric():
  raise ValueError("Input value is not an integer"     )
 return F"""{negative}0b{binary_recursive(int(_lowerCAmelCase     )     )}"""


if __name__ == "__main__":
    from doctest import testmod

    testmod()
