


import gc
import random
import unittest

import numpy as np
import torch
from PIL import Image

from diffusers import (
    DDIMScheduler,
    KandinskyVaaInpaintPipeline,
    KandinskyVaaPriorPipeline,
    UNetaDConditionModel,
    VQModel,
)
from diffusers.utils import floats_tensor, load_image, load_numpy, slow, torch_device
from diffusers.utils.testing_utils import enable_full_determinism, require_torch_gpu

from ..test_pipelines_common import PipelineTesterMixin, assert_mean_pixel_difference


enable_full_determinism()
class 	UpperCAmelCase_					(					UpperCamelCase		, unittest.TestCase    ):

   '''simple docstring'''




   __A :      Union[str, Any]     		=		KandinskyVaaInpaintPipeline
   __A :      Any     		=		["image_embeds", "negative_image_embeds", "image", "mask_image"]
   __A :      Any     		=		[
       "image_embeds",
       "negative_image_embeds",
       "image",
       "mask_image",
   ]
   __A :      Dict     		=		[
       "generator",
       "height",
       "width",
       "latents",
       "guidance_scale",
       "num_inference_steps",
       "return_dict",
       "guidance_scale",
       "num_images_per_prompt",
       "output_type",
       "return_dict",
   ]
   __A :      List[Any]     		=		False






   @property
   def _snake_case    (		self ):






     """simple docstring"""





     return 32






   @property
   def _snake_case    (		self ):






     """simple docstring"""





     return 32






   @property
   def _snake_case    (		self ):






     """simple docstring"""





     return self.time_input_dim






   @property
   def _snake_case    (		self ):






     """simple docstring"""





     return self.time_input_dim * 4






   @property
   def _snake_case    (		self ):






     """simple docstring"""





     return 100






   @property
   def _snake_case    (		self ):






     """simple docstring"""





     torch.manual_seed(0 )

     lowerCamelCase :					Dict    =				{
         "in_channels": 9,
         # Out channels is double in channels because predicts mean and variance
         "out_channels": 8,
         "addition_embed_type": "image",
         "down_block_types": ("ResnetDownsampleBlock2D", "SimpleCrossAttnDownBlock2D"),
         "up_block_types": ("SimpleCrossAttnUpBlock2D", "ResnetUpsampleBlock2D"),
         "mid_block_type": "UNetMidBlock2DSimpleCrossAttn",
         "block_out_channels": (self.block_out_channels_a, self.block_out_channels_a * 2),
         "layers_per_block": 1,
         "encoder_hid_dim": self.text_embedder_hidden_size,
         "encoder_hid_dim_type": "image_proj",
         "cross_attention_dim": self.cross_attention_dim,
         "attention_head_dim": 4,
         "resnet_time_scale_shift": "scale_shift",
         "class_embed_type": None,
     }

     lowerCamelCase :					Any    =				UNetaDConditionModel(**__A )
     return model






   @property
   def _snake_case    (		self ):






     """simple docstring"""





     return {
         "block_out_channels": [32, 64],
         "down_block_types": ["DownEncoderBlock2D", "AttnDownEncoderBlock2D"],
         "in_channels": 3,
         "latent_channels": 4,
         "layers_per_block": 1,
         "norm_num_groups": 8,
         "norm_type": "spatial",
         "num_vq_embeddings": 12,
         "out_channels": 3,
         "up_block_types": [
             "AttnUpDecoderBlock2D",
             "UpDecoderBlock2D",
         ],
         "vq_embed_dim": 4,
     }






   @property
   def _snake_case    (		self ):






     """simple docstring"""





     torch.manual_seed(0 )
     lowerCamelCase :					Any    =				VQModel(**self.dummy_movq_kwargs )
     return model






   def _snake_case    (		self ):






     """simple docstring"""





     lowerCamelCase :					Union[str, Any]    =				self.dummy_unet
     lowerCamelCase :					Any    =				self.dummy_movq

     lowerCamelCase :					Dict    =				DDIMScheduler(
         num_train_timesteps=1000						,				beta_schedule="linear"						,				beta_start=0.00085						,				beta_end=0.012						,				clip_sample=__A						,				set_alpha_to_one=__A						,				steps_offset=1						,				prediction_type="epsilon"						,				thresholding=__A						,				)

     lowerCamelCase :					Optional[Any]    =				{
         "unet": unet,
         "scheduler": scheduler,
         "movq": movq,
     }

     return components






   def _snake_case    (		self						,				__A						,				__A=0 ):






     """simple docstring"""





     lowerCamelCase :					Union[str, Any]    =				floats_tensor((1, self.text_embedder_hidden_size)						,				rng=random.Random(__A ) ).to(__A )
     lowerCamelCase :					Optional[Any]    =				floats_tensor((1, self.text_embedder_hidden_size)						,				rng=random.Random(seed + 1 ) ).to(
         __A )
     # create init_image
     lowerCamelCase :					Any    =				floats_tensor((1, 3, 64, 64)						,				rng=random.Random(__A ) ).to(__A )
     lowerCamelCase :					int    =				image.cpu().permute(0						,				2						,				3						,				1 )[0]
     lowerCamelCase :					int    =				Image.fromarray(np.uinta(__A ) ).convert("RGB" ).resize((256, 256) )
     # create mask
     lowerCamelCase :					int    =				np.ones((64, 64)						,				dtype=np.floataa )
     lowerCamelCase :					Optional[int]    =				0

     if str(__A ).startswith("mps" ):
       lowerCamelCase :					Optional[Any]    =				torch.manual_seed(__A )
     else:
       lowerCamelCase :					List[Any]    =				torch.Generator(device=__A ).manual_seed(__A )
     lowerCamelCase :					Optional[Any]    =				{
         "image": init_image,
         "mask_image": mask,
         "image_embeds": image_embeds,
         "negative_image_embeds": negative_image_embeds,
         "generator": generator,
         "height": 64,
         "width": 64,
         "num_inference_steps": 2,
         "guidance_scale": 4.0,
         "output_type": "np",
     }
     return inputs






   def _snake_case    (		self ):






     """simple docstring"""





     lowerCamelCase :					Any    =				"cpu"

     lowerCamelCase :					List[Any]    =				self.get_dummy_components()

     lowerCamelCase :					str    =				self.pipeline_class(**__A )
     lowerCamelCase :					Tuple    =				pipe.to(__A )

     pipe.set_progress_bar_config(disable=__A )

     lowerCamelCase :					List[str]    =				pipe(**self.get_dummy_inputs(__A ) )
     lowerCamelCase :					List[str]    =				output.images

     lowerCamelCase :					Optional[Any]    =				pipe(
         **self.get_dummy_inputs(__A )						,				return_dict=__A						,				)[0]

     lowerCamelCase :					Dict    =				image[0, -3:, -3:, -1]
     lowerCamelCase :					Union[str, Any]    =				image_from_tuple[0, -3:, -3:, -1]

     print(F"""image.shape {image.shape}""" )

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

     lowerCamelCase :					Optional[int]    =				np.array(
         [0.50775903, 0.49527195, 0.48824543, 0.50192237, 0.48644906, 0.49373814, 0.4780598, 0.47234827, 0.48327848] )

     assert (
         np.abs(image_slice.flatten() - expected_slice ).max() < 1e-2
     ), F""" expected_slice {expected_slice}, but got {image_slice.flatten()}"""
     assert (
         np.abs(image_from_tuple_slice.flatten() - expected_slice ).max() < 1e-2
     ), F""" expected_slice {expected_slice}, but got {image_from_tuple_slice.flatten()}"""






   def _snake_case    (		self ):






     """simple docstring"""





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



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

   '''simple docstring'''




   def _snake_case    (		self ):






     """simple docstring"""





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






   def _snake_case    (		self ):






     """simple docstring"""





     lowerCamelCase :					str    =				load_numpy(
         "https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main"
         "/kandinskyv22/kandinskyv22_inpaint_cat_with_hat_fp16.npy" )

     lowerCamelCase :					List[Any]    =				load_image(
         "https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main" "/kandinsky/cat.png" )
     lowerCamelCase :					Union[str, Any]    =				np.ones((768, 768)						,				dtype=np.floataa )
     lowerCamelCase :					str    =				0

     lowerCamelCase :					Dict    =				"a hat"

     lowerCamelCase :					str    =				KandinskyVaaPriorPipeline.from_pretrained(
         "kandinsky-community/kandinsky-2-2-prior"						,				torch_dtype=torch.floataa )
     pipe_prior.to(__A )

     lowerCamelCase :					Any    =				KandinskyVaaInpaintPipeline.from_pretrained(
         "kandinsky-community/kandinsky-2-2-decoder-inpaint"						,				torch_dtype=torch.floataa )
     lowerCamelCase :					Dict    =				pipeline.to(__A )
     pipeline.set_progress_bar_config(disable=__A )

     lowerCamelCase :					Optional[Any]    =				torch.Generator(device="cpu" ).manual_seed(0 )
     lowerCamelCase    ,	lowerCamelCase :					List[str]    =				pipe_prior(
         __A						,				generator=__A						,				num_inference_steps=5						,				negative_prompt=""						,				).to_tuple()

     lowerCamelCase :					Optional[int]    =				pipeline(
         image=__A						,				mask_image=__A						,				image_embeds=__A						,				negative_image_embeds=__A						,				generator=__A						,				num_inference_steps=100						,				height=768						,				width=768						,				output_type="np"						,				)

     lowerCamelCase :					Any    =				output.images[0]

     assert image.shape == (768, 768, 3)

     assert_mean_pixel_difference(__A						,				__A )







import argparse

import intel_extension_for_pytorch as ipex
import torch

from diffusers import DPMSolverMultistepScheduler, StableDiffusionPipeline


_snake_case             =     argparse.ArgumentParser('''Stable Diffusion script with intel optimization''', add_help=False)
parser.add_argument('''--dpm''', action='''store_true''', help='''Enable DPMSolver or not''')
parser.add_argument('''--steps''', default=None, type=int, help='''Num inference steps''')
_snake_case             =     parser.parse_args()


_snake_case             =     '''cpu'''
_snake_case             =     '''a lovely <dicoo> in red dress and hat, in the snowly and brightly night, with many brighly buildings'''

_snake_case             =     '''path-to-your-trained-model'''
_snake_case             =     StableDiffusionPipeline.from_pretrained(model_id)
if args.dpm:
   _snake_case             =     DPMSolverMultistepScheduler.from_config(pipe.scheduler.config)
_snake_case             =     pipe.to(device)

# to channels last
_snake_case             =     pipe.unet.to(memory_format=torch.channels_last)
_snake_case             =     pipe.vae.to(memory_format=torch.channels_last)
_snake_case             =     pipe.text_encoder.to(memory_format=torch.channels_last)
if pipe.requires_safety_checker:
   _snake_case             =     pipe.safety_checker.to(memory_format=torch.channels_last)

# optimize with ipex
_snake_case             =     torch.randn(2, 4, 64, 64)
_snake_case             =     torch.rand(1) * 9_99
_snake_case             =     torch.randn(2, 77, 7_68)
_snake_case             =     (sample, timestep, encoder_hidden_status)
try:
   _snake_case             =     ipex.optimize(pipe.unet.eval(), dtype=torch.bfloataa, inplace=True, sample_input=input_example)
except Exception:
   _snake_case             =     ipex.optimize(pipe.unet.eval(), dtype=torch.bfloataa, inplace=True)
_snake_case             =     ipex.optimize(pipe.vae.eval(), dtype=torch.bfloataa, inplace=True)
_snake_case             =     ipex.optimize(pipe.text_encoder.eval(), dtype=torch.bfloataa, inplace=True)
if pipe.requires_safety_checker:
   _snake_case             =     ipex.optimize(pipe.safety_checker.eval(), dtype=torch.bfloataa, inplace=True)

# compute
_snake_case             =     6_66
_snake_case             =     torch.Generator(device).manual_seed(seed)
_snake_case             =     {'''generator''': generator}
if args.steps is not None:
   _snake_case             =     args.steps

with torch.cpu.amp.autocast(enabled=True, dtype=torch.bfloataa):
   _snake_case             =     pipe(prompt, **generate_kwargs).images[0]

# save image
image.save('''generated.png''')



