






"""simple docstring"""



import gc
import unittest

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

from diffusers import (
    AutoencoderKL,
    DiffusionPipeline,
    EulerDiscreteScheduler,
    StableDiffusionXLPipeline,
    UNetaDConditionModel,
)
from diffusers.utils import slow, torch_device
from diffusers.utils.testing_utils import enable_full_determinism, 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 PipelineLatentTesterMixin, PipelineTesterMixin


enable_full_determinism()




class        lowerCAmelCase__       (  __magic_name__				,		__magic_name__				,		unittest.TestCase					):
   SCREAMING_SNAKE_CASE_					      =StableDiffusionXLPipeline
   SCREAMING_SNAKE_CASE_					      =TEXT_TO_IMAGE_PARAMS
   SCREAMING_SNAKE_CASE_					      =TEXT_TO_IMAGE_BATCH_PARAMS
   SCREAMING_SNAKE_CASE_					      =TEXT_TO_IMAGE_IMAGE_PARAMS
   SCREAMING_SNAKE_CASE_					      =TEXT_TO_IMAGE_IMAGE_PARAMS
   def 	__a				(						self	:     Union[str, Any]				):


        '''simple docstring'''
        torch.manual_seed(0				)
        UpperCAmelCase__		:  Union[str, Any]				=				UNetaDConditionModel(
            block_out_channels=(3_2, 6_4)				,       layers_per_block=2				,       sample_size=3_2				,       in_channels=4				,       out_channels=4				,       down_block_types=("DownBlock2D", "CrossAttnDownBlock2D")				,       up_block_types=("CrossAttnUpBlock2D", "UpBlock2D")				,       attention_head_dim=(2, 4)				,       use_linear_projection=_UpperCamelCase				,       addition_embed_type="text_time"				,       addition_time_embed_dim=8				,       transformer_layers_per_block=(1, 2)				,       projection_class_embeddings_input_dim=8_0				,       cross_attention_dim=6_4				,       )
        UpperCAmelCase__		:  Optional[int]				=				EulerDiscreteScheduler(
            beta_start=0.0_0085				,       beta_end=0.012				,       steps_offset=1				,       beta_schedule="scaled_linear"				,       timestep_spacing="leading"				,       )
        torch.manual_seed(0				)
        UpperCAmelCase__		:  Any				=				AutoencoderKL(
            block_out_channels=[3_2, 6_4]				,       in_channels=3				,       out_channels=3				,       down_block_types=["DownEncoderBlock2D", "DownEncoderBlock2D"]				,       up_block_types=["UpDecoderBlock2D", "UpDecoderBlock2D"]				,       latent_channels=4				,       sample_size=1_2_8				,       )
        torch.manual_seed(0				)
        UpperCAmelCase__		:  Tuple				=				CLIPTextConfig(
            bos_token_id=0				,       eos_token_id=2				,       hidden_size=3_2				,       intermediate_size=3_7				,       layer_norm_eps=1e-05				,       num_attention_heads=4				,       num_hidden_layers=5				,       pad_token_id=1				,       vocab_size=1_0_0_0				,       hidden_act="gelu"				,       projection_dim=3_2				,       )
        UpperCAmelCase__		:  List[str]				=				CLIPTextModel(_UpperCamelCase				)
        UpperCAmelCase__		:  Optional[Any]				=				CLIPTokenizer.from_pretrained("hf-internal-testing/tiny-random-clip"				,       local_files_only=_UpperCamelCase				)

        UpperCAmelCase__		:  Optional[int]				=				CLIPTextModelWithProjection(_UpperCamelCase				)
        UpperCAmelCase__		:  Optional[int]				=				CLIPTokenizer.from_pretrained("hf-internal-testing/tiny-random-clip"				,       local_files_only=_UpperCamelCase				)

        UpperCAmelCase__		:  Union[str, Any]				=				{
            "unet": unet,
            "scheduler": scheduler,
            "vae": vae,
            "text_encoder": text_encoder,
            "tokenizer": tokenizer,
            "text_encoder_2": text_encoder_a,
            "tokenizer_2": tokenizer_a,
            # "safety_checker": None,
            # "feature_extractor": None,
        }
        return components
   def 	__a				(						self	:     Union[str, Any]				,       snake_case__	:     Optional[Any]				,       snake_case__	:     Any=0				):


        '''simple docstring'''

        if str(_UpperCamelCase				).startswith("mps"				):
             UpperCAmelCase__		:  Any				=				torch.manual_seed(_UpperCamelCase				)
        else:
             UpperCAmelCase__		:  Dict				=				torch.Generator(device=_UpperCamelCase				).manual_seed(_UpperCamelCase				)
        UpperCAmelCase__		:  Optional[Any]				=				{
            "prompt": "A painting of a squirrel eating a burger",
            "generator": generator,
            "num_inference_steps": 2,
            "guidance_scale": 5.0,
            "output_type": "numpy",
        }
        return inputs
   def 	__a				(						self	:     Optional[int]				):


        '''simple docstring'''
        UpperCAmelCase__		:  int				=				"cpu"  # ensure determinism for the device-dependent torch.Generator
        UpperCAmelCase__		:  Any				=				self.get_dummy_components()
        UpperCAmelCase__		:  str				=				StableDiffusionXLPipeline(**_UpperCamelCase				)
        UpperCAmelCase__		:  Any				=				sd_pipe.to(_UpperCamelCase				)
        sd_pipe.set_progress_bar_config(disable=_UpperCamelCase				)

        UpperCAmelCase__		:  List[str]				=				self.get_dummy_inputs(_UpperCamelCase				)
        UpperCAmelCase__		:  Union[str, Any]				=				sd_pipe(**_UpperCamelCase				).images
        UpperCAmelCase__		:  List[Any]				=				image[0, -3:, -3:, -1]

        assert image.shape == (1, 6_4, 6_4, 3)
        UpperCAmelCase__		:  List[str]				=				np.array([0.5873, 0.6128, 0.4797, 0.5122, 0.5674, 0.4639, 0.5227, 0.5149, 0.4747]				)

        assert np.abs(image_slice.flatten() - expected_slice				).max() < 1e-2
   def 	__a				(						self	:     int				):


        '''simple docstring'''
        UpperCAmelCase__		:  Optional[int]				=				self.get_dummy_components()
        UpperCAmelCase__		:  Union[str, Any]				=				StableDiffusionXLPipeline(**_UpperCamelCase				)
        UpperCAmelCase__		:  Optional[int]				=				sd_pipe.to(_UpperCamelCase				)
        UpperCAmelCase__		:  Tuple				=				sd_pipe.to(_UpperCamelCase				)
        sd_pipe.set_progress_bar_config(disable=_UpperCamelCase				)

        # forward without prompt embeds
        UpperCAmelCase__		:  Optional[Any]				=				self.get_dummy_inputs(_UpperCamelCase				)
        UpperCAmelCase__		:  List[str]				=				3 * ["this is a negative prompt"]
        UpperCAmelCase__		:  Tuple				=				negative_prompt
        UpperCAmelCase__		:  Optional[int]				=				3 * [inputs["prompt"]]

        UpperCAmelCase__		:  int				=				sd_pipe(**_UpperCamelCase				)
        UpperCAmelCase__		:  Union[str, Any]				=				output.images[0, -3:, -3:, -1]

        # forward with prompt embeds
        UpperCAmelCase__		:  Tuple				=				self.get_dummy_inputs(_UpperCamelCase				)
        UpperCAmelCase__		:  List[str]				=				3 * ["this is a negative prompt"]
        UpperCAmelCase__		:  int				=				3 * [inputs.pop("prompt"				)]

        (
            (
            UpperCAmelCase__      
        )				,							(
            UpperCAmelCase__      
        )				,							(
            UpperCAmelCase__      
        )				,							(
            UpperCAmelCase__      
        )				,							      
        )		:  List[str]				=				sd_pipe.encode_prompt(_UpperCamelCase				,       negative_prompt=_UpperCamelCase				)

        UpperCAmelCase__		:  str				=				sd_pipe(
            **_UpperCamelCase				,       prompt_embeds=_UpperCamelCase				,       negative_prompt_embeds=_UpperCamelCase				,       pooled_prompt_embeds=_UpperCamelCase				,       negative_pooled_prompt_embeds=_UpperCamelCase				,       )
        UpperCAmelCase__		:  Union[str, Any]				=				output.images[0, -3:, -3:, -1]

        # make sure that it's equal
        assert np.abs(image_slice_a.flatten() - image_slice_a.flatten()				).max() < 1e-4
   def 	__a				(						self	:     Dict				):


        '''simple docstring'''
        super().test_attention_slicing_forward_pass(expected_max_diff=3e-3				)




   def 	__a				(						self	:     List[str]				):


        '''simple docstring'''
        super().test_inference_batch_single_identical(expected_max_diff=3e-3				)




@slow
@require_torch_gpu
class        lowerCAmelCase__       (  unittest.TestCase					):
   def 	__a				(						self	:     Dict				):


        '''simple docstring'''
        super().tearDown()
        gc.collect()
        torch.cuda.empty_cache()
   def 	__a				(						self	:     Dict				,       snake_case__	:     List[Any]				,       snake_case__	:     Optional[Any]="cpu"				,       snake_case__	:     List[Any]=torch.floataa				,       snake_case__	:     str=0				):


        '''simple docstring'''
        UpperCAmelCase__		:  Dict				=				torch.Generator(device=_UpperCamelCase				).manual_seed(_UpperCamelCase				)
        UpperCAmelCase__		:  List[Any]				=				np.random.RandomState(_UpperCamelCase				).standard_normal((1, 4, 6_4, 6_4)				)
        UpperCAmelCase__		:  Dict				=				torch.from_numpy(_UpperCamelCase				).to(device=_UpperCamelCase				,       dtype=_UpperCamelCase				)
        UpperCAmelCase__		:  Tuple				=				{
            "prompt": "a photograph of an astronaut riding a horse",
            "latents": latents,
            "generator": generator,
            "num_inference_steps": 3,
            "guidance_scale": 7.5,
            "output_type": "numpy",
        }
        return inputs




   def 	__a				(						self	:     str				):


        '''simple docstring'''
        UpperCAmelCase__		:  Any				=				DiffusionPipeline.from_pretrained("stabilityai/stable-diffusion-2-base"				)
        pipe.to(_UpperCamelCase				)
        pipe.set_progress_bar_config(disable=_UpperCamelCase				)

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

        assert image.shape == (1, 5_1_2, 5_1_2, 3)
        UpperCAmelCase__		:  Optional[int]				=				np.array([0.4_9493, 0.4_7896, 0.4_0798, 0.5_4214, 0.5_3212, 0.4_8202, 0.4_7656, 0.4_6329, 0.4_8506]				)
        assert np.abs(image_slice - expected_slice				).max() < 7e-3



import gc
import unittest

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

from diffusers import (
    AutoencoderKL,
    DDIMParallelScheduler,
    DDPMParallelScheduler,
    StableDiffusionParadigmsPipeline,
    UNetaDConditionModel,
)
from diffusers.utils import slow, torch_device
from diffusers.utils.testing_utils import (
    enable_full_determinism,
    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 PipelineLatentTesterMixin, PipelineTesterMixin


enable_full_determinism()


class   lowercase			(   a ,				a ,				unittest.TestCase   ):
							lowercase__  :				Optional[Any]									=							StableDiffusionParadigmsPipeline
							lowercase__  :				Optional[Any]									=							TEXT_TO_IMAGE_PARAMS
							lowercase__  :				Union[str, Any]									=							TEXT_TO_IMAGE_BATCH_PARAMS
							lowercase__  :				int									=							TEXT_TO_IMAGE_IMAGE_PARAMS
							lowercase__  :				List[Any]									=							TEXT_TO_IMAGE_IMAGE_PARAMS




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



												'''simple docstring'''

												torch.manual_seed(0					)
												SCREAMING_SNAKE_CASE				=				UNetaDConditionModel(
												    block_out_channels=(32, 64)		,							layers_per_block=2		,							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=_UpperCamelCase		,							)
												SCREAMING_SNAKE_CASE				=				DDIMParallelScheduler(
												    beta_start=0.0_0_0_8_5		,							beta_end=0.0_1_2		,							beta_schedule="scaled_linear"		,							clip_sample=_UpperCamelCase		,							set_alpha_to_one=_UpperCamelCase		,							)
												torch.manual_seed(0					)
												SCREAMING_SNAKE_CASE				=				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					)
												SCREAMING_SNAKE_CASE				=				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=1_000		,							hidden_act="gelu"		,							projection_dim=512		,							)
												SCREAMING_SNAKE_CASE				=				CLIPTextModel(_UpperCamelCase					)
												SCREAMING_SNAKE_CASE				=				CLIPTokenizer.from_pretrained("hf-internal-testing/tiny-random-clip"					)

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




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



												'''simple docstring'''

												if str(_UpperCamelCase					).startswith("mps"					):
																	SCREAMING_SNAKE_CASE				=				torch.manual_seed(_UpperCamelCase					)
												else:
																	SCREAMING_SNAKE_CASE				=				torch.Generator(device=_UpperCamelCase					).manual_seed(_UpperCamelCase					)
												SCREAMING_SNAKE_CASE				=				{
												    "prompt": "a photograph of an astronaut riding a horse",
												    "generator": generator,
												    "num_inference_steps": 10,
												    "guidance_scale": 6.0,
												    "output_type": "numpy",
												    "parallel": 3,
												    "debug": True,
												}
												return inputs




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



												'''simple docstring'''

												SCREAMING_SNAKE_CASE				=				"cpu"  # ensure determinism for the device-dependent torch.Generator
												SCREAMING_SNAKE_CASE				=				self.get_dummy_components()
												SCREAMING_SNAKE_CASE				=				StableDiffusionParadigmsPipeline(**_UpperCamelCase					)
												SCREAMING_SNAKE_CASE				=				sd_pipe.to(_UpperCamelCase					)
												sd_pipe.set_progress_bar_config(disable=_UpperCamelCase					)

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

												SCREAMING_SNAKE_CASE				=				np.array([0.4_7_7_3, 0.5_4_1_7, 0.4_7_2_3, 0.4_9_2_5, 0.5_6_3_1, 0.4_7_5_2, 0.5_2_4_0, 0.4_9_3_5, 0.5_0_2_3]					)

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




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



												'''simple docstring'''

												SCREAMING_SNAKE_CASE				=				"cpu"  # ensure determinism for the device-dependent torch.Generator
												SCREAMING_SNAKE_CASE				=				self.get_dummy_components()
												torch.manual_seed(0					)
												SCREAMING_SNAKE_CASE				=				DDPMParallelScheduler()
												torch.manual_seed(0					)
												SCREAMING_SNAKE_CASE				=				StableDiffusionParadigmsPipeline(**_UpperCamelCase					)
												SCREAMING_SNAKE_CASE				=				sd_pipe.to(_UpperCamelCase					)
												sd_pipe.set_progress_bar_config(disable=_UpperCamelCase					)

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

												SCREAMING_SNAKE_CASE				=				np.array([0.3_5_7_3, 0.4_4_2_0, 0.4_9_6_0, 0.4_7_9_9, 0.3_7_9_6, 0.3_8_7_9, 0.4_8_1_9, 0.4_3_6_5, 0.4_4_6_8]					)

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




							def 		__snake_case(  self      :    Tuple					)    -> int:



												'''simple docstring'''

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




							def 		__snake_case(  self      :    List[Any]					)    -> Union[str, Any]:



												'''simple docstring'''

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





							def 		__snake_case(  self      :    List[Any]					)    -> str:



												'''simple docstring'''

												SCREAMING_SNAKE_CASE				=				"cpu"  # ensure determinism for the device-dependent torch.Generator
												SCREAMING_SNAKE_CASE				=				self.get_dummy_components()
												SCREAMING_SNAKE_CASE				=				StableDiffusionParadigmsPipeline(**_UpperCamelCase					)
												SCREAMING_SNAKE_CASE				=				sd_pipe.to(_UpperCamelCase					)
												sd_pipe.set_progress_bar_config(disable=_UpperCamelCase					)

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

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

												SCREAMING_SNAKE_CASE				=				np.array([0.4_7_7_1, 0.5_4_2_0, 0.4_6_8_3, 0.4_9_1_8, 0.5_6_3_6, 0.4_7_2_5, 0.5_2_3_0, 0.4_9_2_3, 0.5_0_1_5]					)

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



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




							def 		__snake_case(  self      :    List[str]					)    -> List[Any]:



												'''simple docstring'''

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




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



												'''simple docstring'''

												SCREAMING_SNAKE_CASE				=				torch.Generator(device=_UpperCamelCase					).manual_seed(_UpperCamelCase					)
												SCREAMING_SNAKE_CASE				=				{
												    "prompt": "a photograph of an astronaut riding a horse",
												    "generator": generator,
												    "num_inference_steps": 10,
												    "guidance_scale": 7.5,
												    "output_type": "numpy",
												    "parallel": 3,
												    "debug": True,
												}
												return inputs





							def 		__snake_case(  self      :    Dict					)    -> str:



												'''simple docstring'''

												SCREAMING_SNAKE_CASE				=				"stabilityai/stable-diffusion-2-base"
												SCREAMING_SNAKE_CASE				=				DDIMParallelScheduler.from_pretrained(_UpperCamelCase		,							subfolder="scheduler"					)
												SCREAMING_SNAKE_CASE				=				StableDiffusionParadigmsPipeline.from_pretrained(_UpperCamelCase		,							scheduler=_UpperCamelCase		,							safety_checker=_UpperCamelCase					)
												pipe.to(_UpperCamelCase					)
												pipe.set_progress_bar_config(disable=_UpperCamelCase					)
												pipe.enable_attention_slicing()

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

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

												SCREAMING_SNAKE_CASE				=				np.array([0.9_6_2_2, 0.9_6_0_2, 0.9_7_4_8, 0.9_5_9_1, 0.9_6_3_0, 0.9_6_9_1, 0.9_6_6_1, 0.9_6_3_1, 0.9_7_4_1]					)

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




