




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 				__a (			snake_case__							,					snake_case__							,					unittest.TestCase					):


     """simple docstring"""


     _A    :   int     =  StableDiffusionXLPipeline
     _A    :   str     =  TEXT_TO_IMAGE_PARAMS
     _A    :   int     =  TEXT_TO_IMAGE_BATCH_PARAMS
     _A    :   Optional[Any]     =  TEXT_TO_IMAGE_IMAGE_PARAMS
     _A    :   Dict     =  TEXT_TO_IMAGE_IMAGE_PARAMS




     def   __A    (     self      :			Tuple						)		->       Optional[int]:


         '''simple docstring'''





         torch.manual_seed(0						)
         SCREAMING_SNAKE_CASE__       =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    ,)
         SCREAMING_SNAKE_CASE__       =EulerDiscreteScheduler(
             beta_start=0.0_0085    ,beta_end=0.012    ,steps_offset=1    ,beta_schedule="""scaled_linear"""    ,timestep_spacing="""leading"""    ,)
         torch.manual_seed(0						)
         SCREAMING_SNAKE_CASE__       =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						)
         SCREAMING_SNAKE_CASE__       =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    ,)
         SCREAMING_SNAKE_CASE__       =CLIPTextModel(_UpperCamelCase						)
         SCREAMING_SNAKE_CASE__       =CLIPTokenizer.from_pretrained("""hf-internal-testing/tiny-random-clip"""    ,local_files_only=_UpperCamelCase						)

         SCREAMING_SNAKE_CASE__       =CLIPTextModelWithProjection(_UpperCamelCase						)
         SCREAMING_SNAKE_CASE__       =CLIPTokenizer.from_pretrained("""hf-internal-testing/tiny-random-clip"""    ,local_files_only=_UpperCamelCase						)

         SCREAMING_SNAKE_CASE__       ={
             "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      :			Optional[Any]    ,_UpperCamelCase      :			List[Any]    ,_UpperCamelCase      :			str=0						)		->       Any:


         '''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 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      :			Union[str, Any]						)		->       Any:


         '''simple docstring'''





         SCREAMING_SNAKE_CASE__       ="cpu"  # ensure determinism for the device-dependent torch.Generator
         SCREAMING_SNAKE_CASE__       =self.get_dummy_components()
         SCREAMING_SNAKE_CASE__       =StableDiffusionXLPipeline(**_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, 6_4, 6_4, 3)
         SCREAMING_SNAKE_CASE__       =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      :			Tuple						)		->       Any:


         '''simple docstring'''





         SCREAMING_SNAKE_CASE__       =self.get_dummy_components()
         SCREAMING_SNAKE_CASE__       =StableDiffusionXLPipeline(**_UpperCamelCase						)
         SCREAMING_SNAKE_CASE__       =sd_pipe.to(_UpperCamelCase						)
         SCREAMING_SNAKE_CASE__       =sd_pipe.to(_UpperCamelCase						)
         sd_pipe.set_progress_bar_config(disable=_UpperCamelCase						)

         # forward without prompt embeds
         SCREAMING_SNAKE_CASE__       =self.get_dummy_inputs(_UpperCamelCase						)
         SCREAMING_SNAKE_CASE__       =3 * ["this is a negative prompt"]
         SCREAMING_SNAKE_CASE__       =negative_prompt
         SCREAMING_SNAKE_CASE__       =3 * [inputs["prompt"]]

         SCREAMING_SNAKE_CASE__       =sd_pipe(**_UpperCamelCase						)
         SCREAMING_SNAKE_CASE__       =output.images[0, -3:, -3:, -1]

         # forward with prompt embeds
         SCREAMING_SNAKE_CASE__       =self.get_dummy_inputs(_UpperCamelCase						)
         SCREAMING_SNAKE_CASE__       =3 * ["this is a negative prompt"]
         SCREAMING_SNAKE_CASE__       =3 * [inputs.pop("""prompt"""						)]

         (
             SCREAMING_SNAKE_CASE__  
         )       =sd_pipe.encode_prompt(_UpperCamelCase    ,negative_prompt=_UpperCamelCase						)

         SCREAMING_SNAKE_CASE__       =sd_pipe(
             **_UpperCamelCase    ,prompt_embeds=_UpperCamelCase    ,negative_prompt_embeds=_UpperCamelCase    ,pooled_prompt_embeds=_UpperCamelCase    ,negative_pooled_prompt_embeds=_UpperCamelCase    ,)
         SCREAMING_SNAKE_CASE__       =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      :			Optional[int]						)		->       Tuple:


         '''simple docstring'''





         super().test_attention_slicing_forward_pass(expected_max_diff=3e-3						)




     def   __A    (     self      :			Tuple						)		->       Optional[Any]:


         '''simple docstring'''





         super().test_inference_batch_single_identical(expected_max_diff=3e-3						)






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


     """simple docstring"""


     def   __A    (     self      :			Tuple						)		->       Optional[int]:


         '''simple docstring'''





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




     def   __A    (     self      :			Union[str, Any]    ,_UpperCamelCase      :			Tuple    ,_UpperCamelCase      :			List[Any]="cpu"    ,_UpperCamelCase      :			str=torch.floataa    ,_UpperCamelCase      :			Any=0						)		->       Optional[Any]:


         '''simple docstring'''





         SCREAMING_SNAKE_CASE__       =torch.Generator(device=_UpperCamelCase						).manual_seed(_UpperCamelCase						)
         SCREAMING_SNAKE_CASE__       =np.random.RandomState(_UpperCamelCase						).standard_normal((1, 4, 6_4, 6_4)						)
         SCREAMING_SNAKE_CASE__       =torch.from_numpy(_UpperCamelCase						).to(device=_UpperCamelCase    ,dtype=_UpperCamelCase						)
         SCREAMING_SNAKE_CASE__       ={
             "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						)		->       str:


         '''simple docstring'''





         SCREAMING_SNAKE_CASE__       =DiffusionPipeline.from_pretrained("""stabilityai/stable-diffusion-2-base"""						)
         pipe.to(_UpperCamelCase						)
         pipe.set_progress_bar_config(disable=_UpperCamelCase						)

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

         assert image.shape == (1, 5_1_2, 5_1_2, 3)
         SCREAMING_SNAKE_CASE__       =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 tempfile
import unittest

from diffusers import (
    DDIMScheduler,
    DDPMScheduler,
    DPMSolverMultistepScheduler,
    EulerAncestralDiscreteScheduler,
    EulerDiscreteScheduler,
    PNDMScheduler,
    logging,
)
from diffusers.configuration_utils import ConfigMixin, register_to_config
from diffusers.utils.testing_utils import CaptureLogger
class 				__a (			__lowerCamelCase					):


     """simple docstring"""


     _A    :   List[Any]     =  "config.json"




     @register_to_config
     def __init__(     self      :			int    ,_UpperCamelCase      :			int=2    ,_UpperCamelCase      :			Optional[Any]=5    ,_UpperCamelCase      :			List[str]=(2, 5)    ,_UpperCamelCase      :			List[Any]="for diffusion"    ,_UpperCamelCase      :			Optional[int]=[1, 3]    ,)		->       Any:


         '''simple docstring'''





         pass






class 				__a (			__lowerCamelCase					):


     """simple docstring"""


     _A    :   List[str]     =  "config.json"




     @register_to_config
     def __init__(     self      :			List[Any]    ,_UpperCamelCase      :			str=2    ,_UpperCamelCase      :			List[str]=5    ,_UpperCamelCase      :			Optional[Any]=(2, 5)    ,_UpperCamelCase      :			Union[str, Any]="for diffusion"    ,_UpperCamelCase      :			Tuple=[1, 3]    ,)		->       List[str]:


         '''simple docstring'''





         pass






class 				__a (			__lowerCamelCase					):


     """simple docstring"""


     _A    :   Optional[Any]     =  "config.json"




     @register_to_config
     def __init__(     self      :			str    ,_UpperCamelCase      :			Optional[int]=2    ,_UpperCamelCase      :			List[str]=5    ,_UpperCamelCase      :			List[Any]=(2, 5)    ,_UpperCamelCase      :			Dict="for diffusion"    ,_UpperCamelCase      :			Union[str, Any]=[1, 3]    ,_UpperCamelCase      :			Dict=[1, 3]    ,)		->       Any:


         '''simple docstring'''





         pass






class 				__a (			__lowerCamelCase					):


     """simple docstring"""


     _A    :   List[Any]     =  "config.json"




     @register_to_config
     def __init__(     self      :			Union[str, Any]    ,_UpperCamelCase      :			str=2    ,_UpperCamelCase      :			List[str]=5    ,_UpperCamelCase      :			List[Any]=(2, 5)    ,_UpperCamelCase      :			int="for diffusion"    ,_UpperCamelCase      :			Any=[1, 5]    ,_UpperCamelCase      :			Optional[Any]=[5, 4]    ,)		->       List[Any]:


         '''simple docstring'''





         pass






class 				__a (			unittest.TestCase					):


     """simple docstring"""


     def   __A    (     self      :			Dict						)		->       Optional[Any]:


         '''simple docstring'''





         with self.assertRaises(_UpperCamelCase						):
             ConfigMixin.load_config("""dummy_path"""						)




     def   __A    (     self      :			int						)		->       List[str]:


         '''simple docstring'''





         SCREAMING_SNAKE_CASE__       =SampleObject()
         SCREAMING_SNAKE_CASE__       =obj.config
         assert config["a"] == 2
         assert config["b"] == 5
         assert config["c"] == (2, 5)
         assert config["d"] == "for diffusion"
         assert config["e"] == [1, 3]

         # init ignore private arguments
         SCREAMING_SNAKE_CASE__       =SampleObject(_name_or_path="""lalala"""						)
         SCREAMING_SNAKE_CASE__       =obj.config
         assert config["a"] == 2
         assert config["b"] == 5
         assert config["c"] == (2, 5)
         assert config["d"] == "for diffusion"
         assert config["e"] == [1, 3]

         # can override default
         SCREAMING_SNAKE_CASE__       =SampleObject(c=6						)
         SCREAMING_SNAKE_CASE__       =obj.config
         assert config["a"] == 2
         assert config["b"] == 5
         assert config["c"] == 6
         assert config["d"] == "for diffusion"
         assert config["e"] == [1, 3]

         # can use positional arguments.
         SCREAMING_SNAKE_CASE__       =SampleObject(1    ,c=6						)
         SCREAMING_SNAKE_CASE__       =obj.config
         assert config["a"] == 1
         assert config["b"] == 5
         assert config["c"] == 6
         assert config["d"] == "for diffusion"
         assert config["e"] == [1, 3]




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


         '''simple docstring'''





         SCREAMING_SNAKE_CASE__       =SampleObject()
         SCREAMING_SNAKE_CASE__       =obj.config

         assert config["a"] == 2
         assert config["b"] == 5
         assert config["c"] == (2, 5)
         assert config["d"] == "for diffusion"
         assert config["e"] == [1, 3]

         with tempfile.TemporaryDirectory() as tmpdirname:
             obj.save_config(_UpperCamelCase						)
             SCREAMING_SNAKE_CASE__       =SampleObject.from_config(SampleObject.load_config(_UpperCamelCase						)						)
             SCREAMING_SNAKE_CASE__       =new_obj.config

         # unfreeze configs
         SCREAMING_SNAKE_CASE__       =dict(_UpperCamelCase						)
         SCREAMING_SNAKE_CASE__       =dict(_UpperCamelCase						)

         assert config.pop("""c"""						) == (2, 5)  # instantiated as tuple
         assert new_config.pop("""c"""						) == [2, 5]  # saved & loaded as list because of json
         config.pop("""_use_default_values"""						)
         assert config == new_config




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


         '''simple docstring'''





         SCREAMING_SNAKE_CASE__       =logging.get_logger("""diffusers.configuration_utils"""						)
         # 30 for warning
         logger.setLevel(3_0						)

         with CaptureLogger(_UpperCamelCase						) as cap_logger:
             SCREAMING_SNAKE_CASE__       =DDIMScheduler.from_pretrained(
                 """hf-internal-testing/tiny-stable-diffusion-torch"""    ,subfolder="""scheduler"""						)

         assert ddim.__class__ == DDIMScheduler
         # no warning should be thrown
         assert cap_logger.out == ""




     def   __A    (     self      :			Dict						)		->       Optional[Any]:


         '''simple docstring'''





         SCREAMING_SNAKE_CASE__       =logging.get_logger("""diffusers.configuration_utils"""						)
         # 30 for warning
         logger.setLevel(3_0						)

         with CaptureLogger(_UpperCamelCase						) as cap_logger:
             SCREAMING_SNAKE_CASE__       =EulerDiscreteScheduler.from_pretrained(
                 """hf-internal-testing/tiny-stable-diffusion-torch"""    ,subfolder="""scheduler"""						)

         assert euler.__class__ == EulerDiscreteScheduler
         # no warning should be thrown
         assert cap_logger.out == ""




     def   __A    (     self      :			str						)		->       int:


         '''simple docstring'''





         SCREAMING_SNAKE_CASE__       =logging.get_logger("""diffusers.configuration_utils"""						)
         # 30 for warning
         logger.setLevel(3_0						)

         with CaptureLogger(_UpperCamelCase						) as cap_logger:
             SCREAMING_SNAKE_CASE__       =EulerAncestralDiscreteScheduler.from_pretrained(
                 """hf-internal-testing/tiny-stable-diffusion-torch"""    ,subfolder="""scheduler"""						)

         assert euler.__class__ == EulerAncestralDiscreteScheduler
         # no warning should be thrown
         assert cap_logger.out == ""




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


         '''simple docstring'''





         SCREAMING_SNAKE_CASE__       =logging.get_logger("""diffusers.configuration_utils"""						)
         # 30 for warning
         logger.setLevel(3_0						)

         with CaptureLogger(_UpperCamelCase						) as cap_logger:
             SCREAMING_SNAKE_CASE__       =PNDMScheduler.from_pretrained(
                 """hf-internal-testing/tiny-stable-diffusion-torch"""    ,subfolder="""scheduler"""						)

         assert pndm.__class__ == PNDMScheduler
         # no warning should be thrown
         assert cap_logger.out == ""




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


         '''simple docstring'''





         SCREAMING_SNAKE_CASE__       =logging.get_logger("""diffusers.configuration_utils"""						)
         # 30 for warning
         logger.setLevel(3_0						)

         with CaptureLogger(_UpperCamelCase						) as cap_logger:
             SCREAMING_SNAKE_CASE__       =DDPMScheduler.from_pretrained(
                 """hf-internal-testing/tiny-stable-diffusion-torch"""    ,subfolder="""scheduler"""    ,prediction_type="""sample"""    ,beta_end=8    ,)

         with CaptureLogger(_UpperCamelCase						) as cap_logger_a:
             SCREAMING_SNAKE_CASE__       =DDPMScheduler.from_pretrained("""google/ddpm-celebahq-256"""    ,beta_start=8_8						)

         assert ddpm.__class__ == DDPMScheduler
         assert ddpm.config.prediction_type == "sample"
         assert ddpm.config.beta_end == 8
         assert ddpm_a.config.beta_start == 8_8

         # no warning should be thrown
         assert cap_logger.out == ""
         assert cap_logger_a.out == ""




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


         '''simple docstring'''





         SCREAMING_SNAKE_CASE__       =logging.get_logger("""diffusers.configuration_utils"""						)
         # 30 for warning
         logger.setLevel(3_0						)

         with CaptureLogger(_UpperCamelCase						) as cap_logger:
             SCREAMING_SNAKE_CASE__       =DPMSolverMultistepScheduler.from_pretrained(
                 """hf-internal-testing/tiny-stable-diffusion-torch"""    ,subfolder="""scheduler"""						)

         assert dpm.__class__ == DPMSolverMultistepScheduler
         # no warning should be thrown
         assert cap_logger.out == ""




     def   __A    (     self      :			str						)		->       str:


         '''simple docstring'''





         SCREAMING_SNAKE_CASE__       =SampleObject()

         SCREAMING_SNAKE_CASE__       ={k: v for k, v in config.config.items() if not k.startswith("""_"""						)}

         # make sure that default config has all keys in `_use_default_values`
         assert set(config_dict.keys()						) == set(config.config._use_default_values						)

         with tempfile.TemporaryDirectory() as tmpdirname:
             config.save_config(_UpperCamelCase						)

             # now loading it with SampleObject2 should put f into `_use_default_values`
             SCREAMING_SNAKE_CASE__       =SampleObjecta.from_config(_UpperCamelCase						)

             assert "f" in config._use_default_values
             assert config.f == [1, 3]

         # now loading the config, should **NOT** use [1, 3] for `f`, but the default [1, 4] value
         # **BECAUSE** it is part of `config._use_default_values`
         SCREAMING_SNAKE_CASE__       =SampleObjecta.from_config(config.config						)
         assert new_config.f == [5, 4]

         config.config._use_default_values.pop()
         SCREAMING_SNAKE_CASE__       =SampleObjecta.from_config(config.config						)
         assert new_config_a.f == [1, 3]

         # Nevertheless "e" should still be correctly loaded to [1, 3] from SampleObject2 instead of defaulting to [1, 5]
         assert new_config_a.e == [1, 3]






