




import tempfile
import unittest

import torch

from diffusers import UNetaDConditionModel
from diffusers.training_utils import EMAModel
from diffusers.utils.testing_utils import enable_full_determinism, skip_mps, torch_device


enable_full_determinism()



class 			a     (   unittest.TestCase			):

     '''simple docstring'''





     lowerCAmelCase				:       Union[str, Any]   		=  'hf-internal-testing/tiny-stable-diffusion-pipe'
     lowerCAmelCase				:       Optional[int]   		=  1
     lowerCAmelCase				:       Union[str, Any]   		=  77
     lowerCAmelCase				:       int   		=  32
     lowerCAmelCase				:       str   		=  4
     lowerCAmelCase				:       str   		=  64
     lowerCAmelCase				:       Optional[Any]   		=  torch.manual_seed(0			)
     def   lowerCamelCase_   (	self   :			List[Any]					,    __snake_case   :			Optional[int]=0.9_999				):
         UpperCAmelCase_							 =     UNetaDConditionModel.from_pretrained(self.model_id					,    subfolder='''unet'''				)
         UpperCAmelCase_							 =     unet.to(__snake_case				)
         UpperCAmelCase_							 =     EMAModel(unet.parameters()					,    decay=__snake_case					,    model_cls=__snake_case					,    model_config=unet.config				)
         return unet, ema_unet
     def   lowerCamelCase_   (	self   :			int				):
         UpperCAmelCase_							 =     torch.randn(
             self.batch_size					,    self.num_in_channels					,    self.latent_height					,    self.latent_width					,    generator=self.generator				).to(__snake_case				)
         UpperCAmelCase_							 =     torch.randint(0					,    10_00					,    size=(self.batch_size,)					,    generator=self.generator				).to(__snake_case				)
         UpperCAmelCase_							 =     torch.randn(
             self.batch_size					,    self.prompt_length					,    self.text_encoder_hidden_dim					,    generator=self.generator				).to(__snake_case				)
         return noisy_latents, timesteps, encoder_hidden_states
     def   lowerCamelCase_   (	self   :			Union[str, Any]					,    __snake_case   :			List[Any]				):
         UpperCAmelCase_							 =     {}
         for k, param in unet.state_dict().items():
             UpperCAmelCase_							 =     torch.randn_like(__snake_case				) + (param * torch.randn_like(__snake_case				))
             updated_state_dict.update({k: updated_param}				)
         unet.load_state_dict(__snake_case				)
         return unet
     def   lowerCamelCase_   (	self   :			Optional[int]				):
         UpperCAmelCase_ ,							UpperCAmelCase_							 =     self.get_models()
         # Take the first (hypothetical) EMA step.
         ema_unet.step(unet.parameters()				)
         assert ema_unet.optimization_step == 1

         # Take two more.
         for _ in range(2				):
             ema_unet.step(unet.parameters()				)
         assert ema_unet.optimization_step == 3
     def   lowerCamelCase_   (	self   :			Union[str, Any]				):
         UpperCAmelCase_ ,							UpperCAmelCase_							 =     self.get_models()
         # Since the `unet` is not being updated (i.e., backprop'd)
         # there won't be any difference between the `params` of `unet`
         # and `ema_unet` even if we call `ema_unet.step(unet.parameters())`.
         ema_unet.step(unet.parameters()				)
         UpperCAmelCase_							 =     list(unet.parameters()				)
         for s_param, param in zip(ema_unet.shadow_params					,    __snake_case				):
             assert torch.allclose(__snake_case					,    __snake_case				)

         # The above holds true even if we call `ema.step()` multiple times since
         # `unet` params are still not being updated.
         for _ in range(4				):
             ema_unet.step(unet.parameters()				)
         for s_param, param in zip(ema_unet.shadow_params					,    __snake_case				):
             assert torch.allclose(__snake_case					,    __snake_case				)
     def   lowerCamelCase_   (	self   :			Tuple				):
         UpperCAmelCase_ ,							UpperCAmelCase_							 =     self.get_models()
         # Here we simulate the parameter updates for `unet`. Since there might
         # be some parameters which are initialized to zero we take extra care to
         # initialize their values to something non-zero before the multiplication.
         UpperCAmelCase_							 =     self.simulate_backprop(__snake_case				)

         # Take the EMA step.
         ema_unet.step(unet_pseudo_updated_step_one.parameters()				)

         # Now the EMA'd parameters won't be equal to the original model parameters.
         UpperCAmelCase_							 =     list(unet_pseudo_updated_step_one.parameters()				)
         for s_param, param in zip(ema_unet.shadow_params					,    __snake_case				):
             assert ~torch.allclose(__snake_case					,    __snake_case				)

         # Ensure this is the case when we take multiple EMA steps.
         for _ in range(4				):
             ema_unet.step(unet.parameters()				)
         for s_param, param in zip(ema_unet.shadow_params					,    __snake_case				):
             assert ~torch.allclose(__snake_case					,    __snake_case				)
     def   lowerCamelCase_   (	self   :			List[str]				):
         # If we call EMA step after a backpropagation consecutively for two times,
         # the shadow params from those two steps should be different.
         UpperCAmelCase_ ,							UpperCAmelCase_							 =     self.get_models()

         # First backprop + EMA
         UpperCAmelCase_							 =     self.simulate_backprop(__snake_case				)
         ema_unet.step(unet_step_one.parameters()				)
         UpperCAmelCase_							 =     ema_unet.shadow_params

         # Second backprop + EMA
         UpperCAmelCase_							 =     self.simulate_backprop(__snake_case				)
         ema_unet.step(unet_step_two.parameters()				)
         UpperCAmelCase_							 =     ema_unet.shadow_params

         for step_one, step_two in zip(__snake_case					,    __snake_case				):
             assert ~torch.allclose(__snake_case					,    __snake_case				)
     def   lowerCamelCase_   (	self   :			Optional[int]				):
         # If there's no decay even if there are backprops, EMA steps
         # won't take any effect i.e., the shadow params would remain the
         # same.
         UpperCAmelCase_ ,							UpperCAmelCase_							 =     self.get_models(decay=0.0				)
         UpperCAmelCase_							 =     self.simulate_backprop(__snake_case				)
         ema_unet.step(unet_step_one.parameters()				)
         UpperCAmelCase_							 =     ema_unet.shadow_params

         UpperCAmelCase_							 =     self.simulate_backprop(__snake_case				)
         ema_unet.step(unet_step_two.parameters()				)
         UpperCAmelCase_							 =     ema_unet.shadow_params

         for step_one, step_two in zip(__snake_case					,    __snake_case				):
             assert torch.allclose(__snake_case					,    __snake_case				)




     @skip_mps
     def   lowerCamelCase_   (	self   :			Optional[int]				):
         UpperCAmelCase_ ,							UpperCAmelCase_							 =     self.get_models()
         UpperCAmelCase_ ,							UpperCAmelCase_ ,							UpperCAmelCase_							 =     self.get_dummy_inputs()

         with tempfile.TemporaryDirectory() as tmpdir:
             ema_unet.save_pretrained(__snake_case				)
             UpperCAmelCase_							 =     UNetaDConditionModel.from_pretrained(__snake_case					,    model_cls=__snake_case				)
             UpperCAmelCase_							 =     loaded_unet.to(unet.device				)

         # Since no EMA step has been performed the outputs should match.
         UpperCAmelCase_							 =     unet(__snake_case					,    __snake_case					,    __snake_case				).sample
         UpperCAmelCase_							 =     loaded_unet(__snake_case					,    __snake_case					,    __snake_case				).sample

         assert torch.allclose(__snake_case					,    __snake_case					,    atol=1E-4				)






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




