


"""simple docstring"""





import gc
import random
import unittest

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

from diffusers import (
    AutoencoderKL,
    DiffusionPipeline,
    EulerDiscreteScheduler,
    StableDiffusionXLImgaImgPipeline,
    UNetaDConditionModel,
)
from diffusers.utils import floats_tensor, slow, torch_device
from diffusers.utils.testing_utils import enable_full_determinism, require_torch_gpu

from ..pipeline_params import (
    IMAGE_TO_IMAGE_IMAGE_PARAMS,
    TEXT_GUIDED_IMAGE_VARIATION_BATCH_PARAMS,
    TEXT_GUIDED_IMAGE_VARIATION_PARAMS,
)
from ..test_pipelines_common import PipelineLatentTesterMixin, PipelineTesterMixin


enable_full_determinism()



class 						snake_case_(				a__							,    a__							,    unittest.TestCase	):
     __UpperCamelCase     =    StableDiffusionXLImgaImgPipeline
     __UpperCamelCase     =    TEXT_GUIDED_IMAGE_VARIATION_PARAMS - {'''height''', '''width'''}
     __UpperCamelCase     =    PipelineTesterMixin.required_optional_params - {'''latents'''}
     __UpperCamelCase     =    TEXT_GUIDED_IMAGE_VARIATION_BATCH_PARAMS
     __UpperCamelCase     =    IMAGE_TO_IMAGE_IMAGE_PARAMS
     __UpperCamelCase     =    IMAGE_TO_IMAGE_IMAGE_PARAMS






     def    lowerCamelCase__   (		self     :    int ):
           torch.manual_seed(0 )
           lowerCAmelCase    :		str				=      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    , )
           lowerCAmelCase    :		Optional[Any]				=      EulerDiscreteScheduler(
               beta_start=0.00_085    , beta_end=0.012    , steps_offset=1    , beta_schedule='''scaled_linear'''    , timestep_spacing='''leading'''    , )
           torch.manual_seed(0 )
           lowerCAmelCase    :		Dict				=      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 )
           lowerCAmelCase    :		Optional[int]				=      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    , )
           lowerCAmelCase    :		int				=      CLIPTextModel(UpperCamelCase_ )
           lowerCAmelCase    :		Optional[int]				=      CLIPTokenizer.from_pretrained('''hf-internal-testing/tiny-random-clip'''    , local_files_only=UpperCamelCase_ )

           lowerCAmelCase    :		str				=      CLIPTextModelWithProjection(UpperCamelCase_ )
           lowerCAmelCase    :		Union[str, Any]				=      CLIPTokenizer.from_pretrained('''hf-internal-testing/tiny-random-clip'''    , local_files_only=UpperCamelCase_ )

           lowerCAmelCase    :		Optional[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    lowerCamelCase__   (		self     :    Optional[int]    , UpperCamelCase_     :    List[Any]    , UpperCamelCase_     :    int=0 ):
           lowerCAmelCase    :		Union[str, Any]				=      floats_tensor((1, 3, 3_2, 3_2)    , rng=random.Random(UpperCamelCase_ ) ).to(UpperCamelCase_ )
           lowerCAmelCase    :		Optional[Any]				=      image / 2 + 0.5
           if str(UpperCamelCase_ ).startswith('''mps''' ):
                 lowerCAmelCase    :		Any				=      torch.manual_seed(UpperCamelCase_ )
           else:
                 lowerCAmelCase    :		str				=      torch.Generator(device=UpperCamelCase_ ).manual_seed(UpperCamelCase_ )
           lowerCAmelCase    :		List[str]				=      {
               '''prompt''': '''A painting of a squirrel eating a burger''',
               '''image''': image,
               '''generator''': generator,
               '''num_inference_steps''': 2,
               '''guidance_scale''': 5.0,
               '''output_type''': '''numpy''',
               '''strength''': 0.75,
           }
           return inputs






     def    lowerCamelCase__   (		self     :    Dict ):
           lowerCAmelCase    :		Union[str, Any]				=      '''cpu'''  # ensure determinism for the device-dependent torch.Generator
           lowerCAmelCase    :		Optional[Any]				=      self.get_dummy_components()
           lowerCAmelCase    :		Optional[int]				=      StableDiffusionXLImgaImgPipeline(**UpperCamelCase_ )
           lowerCAmelCase    :		Tuple				=      sd_pipe.to(UpperCamelCase_ )
           sd_pipe.set_progress_bar_config(disable=UpperCamelCase_ )

           lowerCAmelCase    :		List[str]				=      self.get_dummy_inputs(UpperCamelCase_ )
           lowerCAmelCase    :		Optional[int]				=      sd_pipe(**UpperCamelCase_ ).images
           lowerCAmelCase    :		Any				=      image[0, -3:, -3:, -1]

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

           lowerCAmelCase    :		Optional[Any]				=      np.array([0.4_656, 0.4_840, 0.4_439, 0.6_698, 0.5_574, 0.4_524, 0.5_799, 0.5_943, 0.5_165] )

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






     def    lowerCamelCase__   (		self     :    int ):
           super().test_attention_slicing_forward_pass(expected_max_diff=3E-3 )






     def    lowerCamelCase__   (		self     :    int ):
           super().test_inference_batch_single_identical(expected_max_diff=3E-3 )






     def    lowerCamelCase__   (		self     :    str ):
           pass






     def    lowerCamelCase__   (		self     :    Optional[int] ):
           lowerCAmelCase    :		List[Any]				=      self.get_dummy_components()
           lowerCAmelCase    :		int				=      StableDiffusionXLImgaImgPipeline(**UpperCamelCase_ )
           lowerCAmelCase    :		int				=      sd_pipe.to(UpperCamelCase_ )
           lowerCAmelCase    :		Any				=      sd_pipe.to(UpperCamelCase_ )
           sd_pipe.set_progress_bar_config(disable=UpperCamelCase_ )

           # forward without prompt embeds
           lowerCAmelCase    :		Tuple				=      self.get_dummy_inputs(UpperCamelCase_ )
           lowerCAmelCase    :		Tuple				=      3 * ['''this is a negative prompt''']
           lowerCAmelCase    :		Dict				=      negative_prompt
           lowerCAmelCase    :		Tuple				=      3 * [inputs['''prompt''']]

           lowerCAmelCase    :		Union[str, Any]				=      sd_pipe(**UpperCamelCase_ )
           lowerCAmelCase    :		str				=      output.images[0, -3:, -3:, -1]

           # forward with prompt embeds
           lowerCAmelCase    :		Union[str, Any]				=      self.get_dummy_inputs(UpperCamelCase_ )
           lowerCAmelCase    :		int				=      3 * ['''this is a negative prompt''']
           lowerCAmelCase    :		Optional[int]				=      3 * [inputs.pop('''prompt''' )]

           (
               (
               lowerCAmelCase
           ),					(
               lowerCAmelCase
           ),					(
               lowerCAmelCase
           ),					(
               lowerCAmelCase
           ),					
           )    :		Tuple				=      sd_pipe.encode_prompt(UpperCamelCase_    , negative_prompt=UpperCamelCase_ )

           lowerCAmelCase    :		Any				=      sd_pipe(
               **UpperCamelCase_    , prompt_embeds=UpperCamelCase_    , negative_prompt_embeds=UpperCamelCase_    , pooled_prompt_embeds=UpperCamelCase_    , negative_pooled_prompt_embeds=UpperCamelCase_    , )
           lowerCAmelCase    :		str				=      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



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






     def    lowerCamelCase__   (		self     :    Any ):
           super().tearDown()
           gc.collect()
           torch.cuda.empty_cache()






     def    lowerCamelCase__   (		self     :    List[str]    , UpperCamelCase_     :    int    , UpperCamelCase_     :    Union[str, Any]="cpu"    , UpperCamelCase_     :    Optional[int]=torch.floataa    , UpperCamelCase_     :    int=0 ):
           lowerCAmelCase    :		Union[str, Any]				=      torch.Generator(device=UpperCamelCase_ ).manual_seed(UpperCamelCase_ )
           lowerCAmelCase    :		Optional[Any]				=      np.random.RandomState(UpperCamelCase_ ).standard_normal((1, 4, 6_4, 6_4) )
           lowerCAmelCase    :		Dict				=      torch.from_numpy(UpperCamelCase_ ).to(device=UpperCamelCase_    , dtype=UpperCamelCase_ )
           lowerCAmelCase    :		str				=      {
               '''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    lowerCamelCase__   (		self     :    Tuple ):
           lowerCAmelCase    :		Union[str, Any]				=      DiffusionPipeline.from_pretrained('''stabilityai/stable-diffusion-2-base''' )
           pipe.to(UpperCamelCase_ )
           pipe.set_progress_bar_config(disable=UpperCamelCase_ )

           lowerCAmelCase    :		List[str]				=      self.get_inputs(UpperCamelCase_ )
           lowerCAmelCase    :		Any				=      pipe(**UpperCamelCase_ ).images
           lowerCAmelCase    :		List[Any]				=      image[0, -3:, -3:, -1].flatten()

           assert image.shape == (1, 5_1_2, 5_1_2, 3)
           lowerCAmelCase    :		Dict				=      np.array([0.49_493, 0.47_896, 0.40_798, 0.54_214, 0.53_212, 0.48_202, 0.47_656, 0.46_329, 0.48_506] )
           assert np.abs(image_slice - expected_slice ).max() < 7E-3






"""simple docstring"""





import inspect
from typing import List, Optional, Tuple, Union

import torch

from ...models import UNetaDModel, VQModel
from ...schedulers import DDIMScheduler
from ...utils import randn_tensor
from ..pipeline_utils import DiffusionPipeline, ImagePipelineOutput



class 						snake_case_(				a__	):






     def __init__(		self     :    int    , UpperCamelCase_     :    VQModel    , UpperCamelCase_     :    UNetaDModel    , UpperCamelCase_     :    DDIMScheduler ):
           super().__init__()
           self.register_modules(vqvae=UpperCamelCase_    , unet=UpperCamelCase_    , scheduler=UpperCamelCase_ )






     @torch.no_grad()
     def __call__(		self     :    Union[str, Any]    , UpperCamelCase_     :    int = 1    , UpperCamelCase_     :    Optional[Union[torch.Generator, List[torch.Generator]]] = None    , UpperCamelCase_     :    float = 0.0    , UpperCamelCase_     :    int = 5_0    , UpperCamelCase_     :    Optional[str] = "pil"    , UpperCamelCase_     :    bool = True    , **UpperCamelCase_     :    Optional[int]    , ):
           lowerCAmelCase    :		Dict				=      randn_tensor(
               (batch_size, self.unet.config.in_channels, self.unet.config.sample_size, self.unet.config.sample_size)    , generator=UpperCamelCase_    , )
           lowerCAmelCase    :		Optional[int]				=      latents.to(self.device )

           # scale the initial noise by the standard deviation required by the scheduler
           lowerCAmelCase    :		List[str]				=      latents * self.scheduler.init_noise_sigma

           self.scheduler.set_timesteps(UpperCamelCase_ )

           # prepare extra kwargs for the scheduler step, since not all schedulers have the same signature
           lowerCAmelCase    :		Any				=      '''eta''' in set(inspect.signature(self.scheduler.step ).parameters.keys() )

           lowerCAmelCase    :		List[str]				=      {}
           if accepts_eta:
                 lowerCAmelCase    :		List[Any]				=      eta

           for t in self.progress_bar(self.scheduler.timesteps ):
                 lowerCAmelCase    :		List[str]				=      self.scheduler.scale_model_input(UpperCamelCase_    , UpperCamelCase_ )
                 # predict the noise residual
                 lowerCAmelCase    :		Tuple				=      self.unet(UpperCamelCase_    , UpperCamelCase_ ).sample
                 # compute the previous noisy sample x_t -> x_t-1
                 lowerCAmelCase    :		Optional[Any]				=      self.scheduler.step(UpperCamelCase_    , UpperCamelCase_    , UpperCamelCase_    , **UpperCamelCase_ ).prev_sample

           # decode the image latents with the VAE
           lowerCAmelCase    :		Dict				=      self.vqvae.decode(UpperCamelCase_ ).sample

           lowerCAmelCase    :		Dict				=      (image / 2 + 0.5).clamp(0    , 1 )
           lowerCAmelCase    :		Dict				=      image.cpu().permute(0    , 2    , 3    , 1 ).numpy()
           if output_type == "pil":
                 lowerCAmelCase    :		List[str]				=      self.numpy_to_pil(UpperCamelCase_ )

           if not return_dict:
                 return (image,)

           return ImagePipelineOutput(images=UpperCamelCase_ )


