

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   __lowercase				( UpperCAmelCase_		,       UpperCAmelCase_		,       unittest.TestCase	):




			"""simple docstring"""







			_UpperCAmelCase   :							int										=     StableDiffusionPanoramaPipeline
			_UpperCAmelCase   :							str										=     TEXT_TO_IMAGE_PARAMS
			_UpperCAmelCase   :							Optional[int]										=     TEXT_TO_IMAGE_BATCH_PARAMS
			_UpperCAmelCase   :							str										=     TEXT_TO_IMAGE_IMAGE_PARAMS
			_UpperCAmelCase   :							List[str]										=     TEXT_TO_IMAGE_IMAGE_PARAMS



			def        _SCREAMING_SNAKE_CASE			( self  :     Union[str, Any]):
									torch.manual_seed(0)
									SCREAMING_SNAKE_CASE_: int  		=       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   ,						)
									SCREAMING_SNAKE_CASE_: Union[str, Any]  		=       DDIMScheduler()
									torch.manual_seed(0)
									SCREAMING_SNAKE_CASE_: Union[str, Any]  		=       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)
									SCREAMING_SNAKE_CASE_: Any  		=       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   ,						)
									SCREAMING_SNAKE_CASE_: str  		=       CLIPTextModel(lowerCAmelCase__)
									SCREAMING_SNAKE_CASE_: List[Any]  		=       CLIPTokenizer.from_pretrained("hf-internal-testing/tiny-random-clip")

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



			def        _SCREAMING_SNAKE_CASE			( self  :     Dict   ,						lowerCAmelCase__  :     Optional[Any]   ,						lowerCAmelCase__  :     Optional[int]=0):
									SCREAMING_SNAKE_CASE_: Optional[int]  		=       torch.manual_seed(lowerCAmelCase__)
									SCREAMING_SNAKE_CASE_: List[str]  		=       {
									    "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        _SCREAMING_SNAKE_CASE			( self  :     int):
									SCREAMING_SNAKE_CASE_: Optional[Any]  		=       "cpu"  # ensure determinism for the device-dependent torch.Generator
									SCREAMING_SNAKE_CASE_: Tuple  		=       self.get_dummy_components()
									SCREAMING_SNAKE_CASE_: Tuple  		=       StableDiffusionPanoramaPipeline(**lowerCAmelCase__)
									SCREAMING_SNAKE_CASE_: Dict  		=       sd_pipe.to(lowerCAmelCase__)
									sd_pipe.set_progress_bar_config(disable=lowerCAmelCase__)

									SCREAMING_SNAKE_CASE_: int  		=       self.get_dummy_inputs(lowerCAmelCase__)
									SCREAMING_SNAKE_CASE_: Optional[Any]  		=       sd_pipe(**lowerCAmelCase__).images
									SCREAMING_SNAKE_CASE_: List[Any]  		=       image[0, -3:, -3:, -1]
									assert image.shape == (1, 64, 64, 3)

									SCREAMING_SNAKE_CASE_: Any  		=       np.array([0.6186, 0.5374, 0.4915, 0.4135, 0.4114, 0.4563, 0.5128, 0.4977, 0.4757])

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



			def        _SCREAMING_SNAKE_CASE			( self  :     Tuple):
									super().test_inference_batch_consistent(batch_sizes=[1, 2])



			def        _SCREAMING_SNAKE_CASE			( self  :     List[str]):
									super().test_inference_batch_single_identical(batch_size=2   ,						expected_max_diff=3.25E-3)



			def        _SCREAMING_SNAKE_CASE			( self  :     Any):
									SCREAMING_SNAKE_CASE_: Optional[Any]  		=       "cpu"  # ensure determinism for the device-dependent torch.Generator
									SCREAMING_SNAKE_CASE_: str  		=       self.get_dummy_components()
									SCREAMING_SNAKE_CASE_: Union[str, Any]  		=       StableDiffusionPanoramaPipeline(**lowerCAmelCase__)
									SCREAMING_SNAKE_CASE_: Optional[Any]  		=       sd_pipe.to(lowerCAmelCase__)
									sd_pipe.set_progress_bar_config(disable=lowerCAmelCase__)

									SCREAMING_SNAKE_CASE_: Union[str, Any]  		=       self.get_dummy_inputs(lowerCAmelCase__)
									SCREAMING_SNAKE_CASE_: str  		=       "french fries"
									SCREAMING_SNAKE_CASE_: List[str]  		=       sd_pipe(**lowerCAmelCase__   ,						negative_prompt=lowerCAmelCase__)
									SCREAMING_SNAKE_CASE_: Tuple  		=       output.images
									SCREAMING_SNAKE_CASE_: int  		=       image[0, -3:, -3:, -1]

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

									SCREAMING_SNAKE_CASE_: int  		=       np.array([0.6187, 0.5375, 0.4915, 0.4136, 0.4114, 0.4563, 0.5128, 0.4976, 0.4757])

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



			def        _SCREAMING_SNAKE_CASE			( self  :     Dict):
									SCREAMING_SNAKE_CASE_: Tuple  		=       "cpu"  # ensure determinism for the device-dependent torch.Generator
									SCREAMING_SNAKE_CASE_: List[str]  		=       self.get_dummy_components()
									SCREAMING_SNAKE_CASE_: Any  		=       StableDiffusionPanoramaPipeline(**lowerCAmelCase__)
									SCREAMING_SNAKE_CASE_: List[str]  		=       sd_pipe.to(lowerCAmelCase__)
									sd_pipe.set_progress_bar_config(disable=lowerCAmelCase__)

									SCREAMING_SNAKE_CASE_: str  		=       self.get_dummy_inputs(lowerCAmelCase__)
									SCREAMING_SNAKE_CASE_: Dict  		=       sd_pipe(**lowerCAmelCase__   ,						view_batch_size=2)
									SCREAMING_SNAKE_CASE_: List[Any]  		=       output.images
									SCREAMING_SNAKE_CASE_: int  		=       image[0, -3:, -3:, -1]

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

									SCREAMING_SNAKE_CASE_: Any  		=       np.array([0.6187, 0.5375, 0.4915, 0.4136, 0.4114, 0.4563, 0.5128, 0.4976, 0.4757])

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



			def        _SCREAMING_SNAKE_CASE			( self  :     Union[str, Any]):
									SCREAMING_SNAKE_CASE_: Optional[Any]  		=       "cpu"  # ensure determinism for the device-dependent torch.Generator
									SCREAMING_SNAKE_CASE_: str  		=       self.get_dummy_components()
									SCREAMING_SNAKE_CASE_: Any  		=       EulerAncestralDiscreteScheduler(
									    beta_start=0.0_0085   ,						beta_end=0.012   ,						beta_schedule="scaled_linear")
									SCREAMING_SNAKE_CASE_: Tuple  		=       StableDiffusionPanoramaPipeline(**lowerCAmelCase__)
									SCREAMING_SNAKE_CASE_: List[Any]  		=       sd_pipe.to(lowerCAmelCase__)
									sd_pipe.set_progress_bar_config(disable=lowerCAmelCase__)

									SCREAMING_SNAKE_CASE_: int  		=       self.get_dummy_inputs(lowerCAmelCase__)
									SCREAMING_SNAKE_CASE_: List[Any]  		=       sd_pipe(**lowerCAmelCase__).images
									SCREAMING_SNAKE_CASE_: int  		=       image[0, -3:, -3:, -1]

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

									SCREAMING_SNAKE_CASE_: List[Any]  		=       np.array([0.4024, 0.6510, 0.4901, 0.5378, 0.5813, 0.5622, 0.4795, 0.4467, 0.4952])

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




			def        _SCREAMING_SNAKE_CASE			( self  :     str):
									SCREAMING_SNAKE_CASE_: List[str]  		=       "cpu"  # ensure determinism for the device-dependent torch.Generator
									SCREAMING_SNAKE_CASE_: str  		=       self.get_dummy_components()
									SCREAMING_SNAKE_CASE_: Optional[int]  		=       PNDMScheduler(
									    beta_start=0.0_0085   ,						beta_end=0.012   ,						beta_schedule="scaled_linear"   ,						skip_prk_steps=lowerCAmelCase__)
									SCREAMING_SNAKE_CASE_: Union[str, Any]  		=       StableDiffusionPanoramaPipeline(**lowerCAmelCase__)
									SCREAMING_SNAKE_CASE_: str  		=       sd_pipe.to(lowerCAmelCase__)
									sd_pipe.set_progress_bar_config(disable=lowerCAmelCase__)

									SCREAMING_SNAKE_CASE_: List[Any]  		=       self.get_dummy_inputs(lowerCAmelCase__)
									SCREAMING_SNAKE_CASE_: Tuple  		=       sd_pipe(**lowerCAmelCase__).images
									SCREAMING_SNAKE_CASE_: List[str]  		=       image[0, -3:, -3:, -1]

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

									SCREAMING_SNAKE_CASE_: Optional[int]  		=       np.array([0.6391, 0.6291, 0.4861, 0.5134, 0.5552, 0.4578, 0.5032, 0.5023, 0.4539])

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







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




			"""simple docstring"""







			def        _SCREAMING_SNAKE_CASE			( self  :     List[str]):
									super().tearDown()
									gc.collect()
									torch.cuda.empty_cache()



			def        _SCREAMING_SNAKE_CASE			( self  :     Any   ,						lowerCAmelCase__  :     Optional[Any]=0):
									SCREAMING_SNAKE_CASE_: Union[str, Any]  		=       torch.manual_seed(lowerCAmelCase__)
									SCREAMING_SNAKE_CASE_: int  		=       {
									    "prompt": "a photo of the dolomites",
									    "generator": generator,
									    "num_inference_steps": 3,
									    "guidance_scale": 7.5,
									    "output_type": "numpy",
									}
									return inputs



			def        _SCREAMING_SNAKE_CASE			( self  :     str):
									SCREAMING_SNAKE_CASE_: List[Any]  		=       "stabilityai/stable-diffusion-2-base"
									SCREAMING_SNAKE_CASE_: Any  		=       DDIMScheduler.from_pretrained(lowerCAmelCase__   ,						subfolder="scheduler")
									SCREAMING_SNAKE_CASE_: int  		=       StableDiffusionPanoramaPipeline.from_pretrained(lowerCAmelCase__   ,						scheduler=lowerCAmelCase__   ,						safety_checker=lowerCAmelCase__)
									pipe.to(lowerCAmelCase__)
									pipe.set_progress_bar_config(disable=lowerCAmelCase__)
									pipe.enable_attention_slicing()

									SCREAMING_SNAKE_CASE_: List[str]  		=       self.get_inputs()
									SCREAMING_SNAKE_CASE_: Tuple  		=       pipe(**lowerCAmelCase__).images
									SCREAMING_SNAKE_CASE_: int  		=       image[0, -3:, -3:, -1].flatten()

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

									SCREAMING_SNAKE_CASE_: List[str]  		=       np.array(
									    [
									        0.3696_8392,
									        0.2702_5372,
									        0.3244_6766,
									        0.2837_9387,
									        0.3636_3274,
									        0.3073_3347,
									        0.2710_0027,
									        0.2705_4125,
									        0.2553_6096,
									    ])

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



			def        _SCREAMING_SNAKE_CASE			( self  :     Tuple):
									SCREAMING_SNAKE_CASE_: Dict  		=       StableDiffusionPanoramaPipeline.from_pretrained(
									    "stabilityai/stable-diffusion-2-base"   ,						safety_checker=lowerCAmelCase__)
									SCREAMING_SNAKE_CASE_: str  		=       LMSDiscreteScheduler.from_config(pipe.scheduler.config)
									pipe.to(lowerCAmelCase__)
									pipe.set_progress_bar_config(disable=lowerCAmelCase__)
									pipe.enable_attention_slicing()

									SCREAMING_SNAKE_CASE_: Optional[int]  		=       self.get_inputs()
									SCREAMING_SNAKE_CASE_: List[str]  		=       pipe(**lowerCAmelCase__).images
									SCREAMING_SNAKE_CASE_: Optional[Any]  		=       image[0, -3:, -3:, -1].flatten()

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

									SCREAMING_SNAKE_CASE_: int  		=       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        _SCREAMING_SNAKE_CASE			( self  :     Dict):
									SCREAMING_SNAKE_CASE_: Tuple  		=       0

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

																					SCREAMING_SNAKE_CASE_: Any  		=       np.array(
																					    [
																					        0.1868_1869,
																					        0.3390_7816,
																					        0.536_1276,
																					        0.1443_2865,
																					        -0.0285_6611,
																					        -0.7394_1123,
																					        0.2339_7987,
																					        0.4732_2682,
																					        -0.3782_3164,
																					    ])
																					assert np.abs(latents_slice.flatten() - expected_slice).max() < 5E-2
															elif step == 2:
																					SCREAMING_SNAKE_CASE_: str  		=       latents.detach().cpu().numpy()
																					assert latents.shape == (1, 4, 64, 256)
																					SCREAMING_SNAKE_CASE_: Dict  		=       latents[0, -3:, -3:, -1]

																					SCREAMING_SNAKE_CASE_: List[str]  		=       np.array(
																					    [
																					        0.1853_9645,
																					        0.3398_7248,
																					        0.537_8559,
																					        0.1443_7142,
																					        -0.0245_5261,
																					        -0.733_8317,
																					        0.2399_0755,
																					        0.4735_6272,
																					        -0.378_6505,
																					    ])

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

									SCREAMING_SNAKE_CASE_: Optional[int]  		=       False

									SCREAMING_SNAKE_CASE_: Optional[int]  		=       "stabilityai/stable-diffusion-2-base"
									SCREAMING_SNAKE_CASE_: Dict  		=       DDIMScheduler.from_pretrained(lowerCAmelCase__   ,						subfolder="scheduler")
									SCREAMING_SNAKE_CASE_: List[str]  		=       StableDiffusionPanoramaPipeline.from_pretrained(lowerCAmelCase__   ,						scheduler=lowerCAmelCase__   ,						safety_checker=lowerCAmelCase__)
									SCREAMING_SNAKE_CASE_: List[Any]  		=       pipe.to(lowerCAmelCase__)
									pipe.set_progress_bar_config(disable=lowerCAmelCase__)
									pipe.enable_attention_slicing()

									SCREAMING_SNAKE_CASE_: List[Any]  		=       self.get_inputs()
									pipe(**lowerCAmelCase__   ,						callback=lowerCAmelCase__   ,						callback_steps=1)
									assert callback_fn.has_been_called
									assert number_of_steps == 3




			def        _SCREAMING_SNAKE_CASE			( self  :     Dict):
									torch.cuda.empty_cache()
									torch.cuda.reset_max_memory_allocated()
									torch.cuda.reset_peak_memory_stats()

									SCREAMING_SNAKE_CASE_: Optional[Any]  		=       "stabilityai/stable-diffusion-2-base"
									SCREAMING_SNAKE_CASE_: Optional[Any]  		=       DDIMScheduler.from_pretrained(lowerCAmelCase__   ,						subfolder="scheduler")
									SCREAMING_SNAKE_CASE_: Dict  		=       StableDiffusionPanoramaPipeline.from_pretrained(lowerCAmelCase__   ,						scheduler=lowerCAmelCase__   ,						safety_checker=lowerCAmelCase__)
									SCREAMING_SNAKE_CASE_: int  		=       pipe.to(lowerCAmelCase__)
									pipe.set_progress_bar_config(disable=lowerCAmelCase__)
									pipe.enable_attention_slicing(1)
									pipe.enable_sequential_cpu_offload()

									SCREAMING_SNAKE_CASE_: Union[str, Any]  		=       self.get_inputs()
									SCREAMING_SNAKE_CASE_: int  		=       pipe(**lowerCAmelCase__)

									SCREAMING_SNAKE_CASE_: Union[str, Any]  		=       torch.cuda.max_memory_allocated()
									# make sure that less than 5.2 GB is allocated
									assert mem_bytes < 5.5 * 10**9









from ...utils import (
    OptionalDependencyNotAvailable,
    is_torch_available,
    is_transformers_available,
    is_transformers_version,
)


try:
						if not (is_transformers_available() and is_torch_available() and is_transformers_version(""">=""", """4.25.0""")):
												raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
						from ...utils.dummy_torch_and_transformers_objects import UnCLIPImageVariationPipeline, UnCLIPPipeline
else:
						from .pipeline_unclip import UnCLIPPipeline
						from .pipeline_unclip_image_variation import UnCLIPImageVariationPipeline
						from .text_proj import UnCLIPTextProjModel






