

"""simple docstring"""




import gc
import math
import unittest

import torch

from diffusers import UNetaDModel
from diffusers.utils import floats_tensor, logging, slow, torch_all_close, torch_device
from diffusers.utils.testing_utils import enable_full_determinism

from .test_modeling_common import ModelTesterMixin, UNetTesterMixin


a		=logging.get_logger(__name__)

enable_full_determinism()






class 	A_		(	lowerCamelCase_      ,							lowerCamelCase_      ,							unittest.TestCase    ):
							_UpperCAmelCase   :  str										=				UNetaDModel
							_UpperCAmelCase   :  str										=				'''sample'''
							@property
							def    lowerCAmelCase					(		self		:      Optional[int]):
								__lowerCamelCase    :    List[Any] 	=			4
								__lowerCamelCase    :    str 	=			3
								__lowerCamelCase    :    int 	=			(3_2, 3_2)

								__lowerCamelCase    :    int 	=			floats_tensor((batch_size, num_channels) + sizes).to(__snake_case)
								__lowerCamelCase    :    Any 	=			torch.tensor([1_0]).to(__snake_case)

								return {"sample": noise, "timestep": time_step}
							@property
							def    lowerCAmelCase					(		self		:      Tuple):
								return (3, 3_2, 3_2)
							@property
							def    lowerCAmelCase					(		self		:      List[str]):
								return (3, 3_2, 3_2)


							def    lowerCAmelCase					(		self		:      str):
								__lowerCamelCase    :    Any 	=			{
								    'block_out_channels': (3_2, 6_4),
								    'down_block_types': ('DownBlock2D', 'AttnDownBlock2D'),
								    'up_block_types': ('AttnUpBlock2D', 'UpBlock2D'),
								    'attention_head_dim': 3,
								    'out_channels': 3,
								    'in_channels': 3,
								    'layers_per_block': 2,
								    'sample_size': 3_2,
								}
								__lowerCamelCase    :    Optional[Any] 	=			self.dummy_input
								return init_dict, inputs_dict






class 	A_		(	lowerCamelCase_      ,							lowerCamelCase_      ,							unittest.TestCase    ):
							_UpperCAmelCase   :  int										=				UNetaDModel
							_UpperCAmelCase   :  Any										=				'''sample'''
							@property
							def    lowerCAmelCase					(		self		:      Dict):
								__lowerCamelCase    :    Optional[int] 	=			4
								__lowerCamelCase    :    Optional[Any] 	=			4
								__lowerCamelCase    :    Any 	=			(3_2, 3_2)

								__lowerCamelCase    :    Union[str, Any] 	=			floats_tensor((batch_size, num_channels) + sizes).to(__snake_case)
								__lowerCamelCase    :    Any 	=			torch.tensor([1_0]).to(__snake_case)

								return {"sample": noise, "timestep": time_step}
							@property
							def    lowerCAmelCase					(		self		:      Any):
								return (4, 3_2, 3_2)
							@property
							def    lowerCAmelCase					(		self		:      Any):
								return (4, 3_2, 3_2)
							def    lowerCAmelCase					(		self		:      int):
								__lowerCamelCase    :    List[Any] 	=			{
								    'sample_size': 3_2,
								    'in_channels': 4,
								    'out_channels': 4,
								    'layers_per_block': 2,
								    'block_out_channels': (3_2, 6_4),
								    'attention_head_dim': 3_2,
								    'down_block_types': ('DownBlock2D', 'DownBlock2D'),
								    'up_block_types': ('UpBlock2D', 'UpBlock2D'),
								}
								__lowerCamelCase    :    Tuple 	=			self.dummy_input
								return init_dict, inputs_dict
							def    lowerCAmelCase					(		self		:      str):
								__lowerCamelCase	,			__lowerCamelCase    :    Union[str, Any] 	=			UNetaDModel.from_pretrained('fusing/unet-ldm-dummy-update'       ,output_loading_info=__snake_case)

								self.assertIsNotNone(__snake_case)
								self.assertEqual(len(loading_info['missing_keys'])       ,0)

								model.to(__snake_case)
								__lowerCamelCase    :    Union[str, Any] 	=			model(**self.dummy_input).sample

								assert image is not None, "Make sure output is not None"
							@unittest.skipIf(torch_device != 'cuda'       ,'This test is supposed to run on GPU')
							def    lowerCAmelCase					(		self		:      Tuple):
								__lowerCamelCase	,			__lowerCamelCase    :    List[str] 	=			UNetaDModel.from_pretrained('fusing/unet-ldm-dummy-update'       ,output_loading_info=__snake_case)
								model.to(__snake_case)
								__lowerCamelCase    :    int 	=			model(**self.dummy_input).sample

								assert image is not None, "Make sure output is not None"
							@unittest.skipIf(torch_device != 'cuda'       ,'This test is supposed to run on GPU')
							def    lowerCAmelCase					(		self		:      Union[str, Any]):
								# by defautl model loading will use accelerate as `low_cpu_mem_usage=True`
								__lowerCamelCase	,			__lowerCamelCase    :    List[Any] 	=			UNetaDModel.from_pretrained('fusing/unet-ldm-dummy-update'       ,output_loading_info=__snake_case)
								model_accelerate.to(__snake_case)
								model_accelerate.eval()

								__lowerCamelCase    :    List[Any] 	=			torch.randn(
								    1       ,model_accelerate.config.in_channels       ,model_accelerate.config.sample_size       ,model_accelerate.config.sample_size       ,generator=torch.manual_seed(0)       ,)
								__lowerCamelCase    :    Optional[int] 	=			noise.to(__snake_case)
								__lowerCamelCase    :    Tuple 	=			torch.tensor([1_0] * noise.shape[0]).to(__snake_case)

								__lowerCamelCase    :    List[Any] 	=			model_accelerate(__snake_case       ,__snake_case)['sample']

								# two models don't need to stay in the device at the same time
								del model_accelerate
								torch.cuda.empty_cache()
								gc.collect()

								__lowerCamelCase	,			__lowerCamelCase    :    int 	=			UNetaDModel.from_pretrained(
								    'fusing/unet-ldm-dummy-update'       ,output_loading_info=__snake_case       ,low_cpu_mem_usage=__snake_case)
								model_normal_load.to(__snake_case)
								model_normal_load.eval()
								__lowerCamelCase    :    Tuple 	=			model_normal_load(__snake_case       ,__snake_case)['sample']

								assert torch_all_close(__snake_case       ,__snake_case       ,rtol=1E-3)


							def    lowerCAmelCase					(		self		:      str):
								__lowerCamelCase    :    int 	=			UNetaDModel.from_pretrained('fusing/unet-ldm-dummy-update')
								model.eval()
								model.to(__snake_case)

								__lowerCamelCase    :    Optional[int] 	=			torch.randn(
								    1       ,model.config.in_channels       ,model.config.sample_size       ,model.config.sample_size       ,generator=torch.manual_seed(0)       ,)
								__lowerCamelCase    :    Optional[int] 	=			noise.to(__snake_case)
								__lowerCamelCase    :    List[str] 	=			torch.tensor([1_0] * noise.shape[0]).to(__snake_case)

								with torch.no_grad():
									__lowerCamelCase    :    Union[str, Any] 	=			model(__snake_case       ,__snake_case).sample

								__lowerCamelCase    :    Any 	=			output[0, -1, -3:, -3:].flatten().cpu()
								# fmt: off
								__lowerCamelCase    :    Any 	=			torch.tensor([-13.3258, -20.1100, -15.9873, -17.6617, -23.0596, -17.9419, -13.3675, -16.1889, -12.3800])
								# fmt: on

								self.assertTrue(torch_all_close(__snake_case       ,__snake_case       ,rtol=1E-3))






class 	A_		(	lowerCamelCase_      ,							lowerCamelCase_      ,							unittest.TestCase    ):
							_UpperCAmelCase   :  Dict										=				UNetaDModel
							_UpperCAmelCase   :  Optional[Any]										=				'''sample'''
							@property
							def    lowerCAmelCase					(		self		:      Optional[Any]       ,SCREAMING_SNAKE_CASE__		:      List[Any]=(3_2, 3_2)):
								__lowerCamelCase    :    Optional[int] 	=			4
								__lowerCamelCase    :    List[Any] 	=			3

								__lowerCamelCase    :    str 	=			floats_tensor((batch_size, num_channels) + sizes).to(__snake_case)
								__lowerCamelCase    :    str 	=			torch.tensor(batch_size * [1_0]).to(dtype=torch.intaa       ,device=__snake_case)

								return {"sample": noise, "timestep": time_step}
							@property
							def    lowerCAmelCase					(		self		:      Tuple):
								return (3, 3_2, 3_2)
							@property
							def    lowerCAmelCase					(		self		:      Optional[Any]):
								return (3, 3_2, 3_2)
							def    lowerCAmelCase					(		self		:      Optional[Any]):
								__lowerCamelCase    :    Dict 	=			{
								    'block_out_channels': [3_2, 6_4, 6_4, 6_4],
								    'in_channels': 3,
								    'layers_per_block': 1,
								    'out_channels': 3,
								    'time_embedding_type': 'fourier',
								    'norm_eps': 1E-6,
								    'mid_block_scale_factor': math.sqrt(2.0),
								    'norm_num_groups': None,
								    'down_block_types': [
								        'SkipDownBlock2D',
								        'AttnSkipDownBlock2D',
								        'SkipDownBlock2D',
								        'SkipDownBlock2D',
								    ],
								    'up_block_types': [
								        'SkipUpBlock2D',
								        'SkipUpBlock2D',
								        'AttnSkipUpBlock2D',
								        'SkipUpBlock2D',
								    ],
								}
								__lowerCamelCase    :    str 	=			self.dummy_input
								return init_dict, inputs_dict
							@slow
							def    lowerCAmelCase					(		self		:      str):
								__lowerCamelCase	,			__lowerCamelCase    :    Tuple 	=			UNetaDModel.from_pretrained('google/ncsnpp-celebahq-256'       ,output_loading_info=__snake_case)
								self.assertIsNotNone(__snake_case)
								self.assertEqual(len(loading_info['missing_keys'])       ,0)

								model.to(__snake_case)
								__lowerCamelCase    :    List[str] 	=			self.dummy_input
								__lowerCamelCase    :    Any 	=			floats_tensor((4, 3) + (2_5_6, 2_5_6)).to(__snake_case)
								__lowerCamelCase    :    Optional[int] 	=			noise
								__lowerCamelCase    :    str 	=			model(**__snake_case)

								assert image is not None, "Make sure output is not None"
							@slow
							def    lowerCAmelCase					(		self		:      Union[str, Any]):
								__lowerCamelCase    :    List[Any] 	=			UNetaDModel.from_pretrained('google/ncsnpp-celebahq-256')
								model.to(__snake_case)

								__lowerCamelCase    :    Dict 	=			4
								__lowerCamelCase    :    Optional[int] 	=			3
								__lowerCamelCase    :    Union[str, Any] 	=			(2_5_6, 2_5_6)

								__lowerCamelCase    :    Tuple 	=			torch.ones((batch_size, num_channels) + sizes).to(__snake_case)
								__lowerCamelCase    :    Tuple 	=			torch.tensor(batch_size * [1E-4]).to(__snake_case)

								with torch.no_grad():
									__lowerCamelCase    :    Optional[Any] 	=			model(__snake_case       ,__snake_case).sample

								__lowerCamelCase    :    Optional[int] 	=			output[0, -3:, -3:, -1].flatten().cpu()
								# fmt: off
								__lowerCamelCase    :    Any 	=			torch.tensor([-4_8_4_2.8_6_9_1, -6_4_9_9.6_6_3_1, -3_8_0_0.1_9_5_3, -7_9_7_8.2_6_8_6, -1_0_9_8_0.7_1_2_9, -2_0_0_2_8.8_5_3_5, 8_1_4_8.2_8_2_2, 2_3_4_2.2_9_0_5, 5_6_7.7_6_0_8])
								# fmt: on

								self.assertTrue(torch_all_close(__snake_case       ,__snake_case       ,rtol=1E-2))
							def    lowerCAmelCase					(		self		:      Dict):
								__lowerCamelCase    :    Optional[Any] 	=			UNetaDModel.from_pretrained('fusing/ncsnpp-ffhq-ve-dummy-update')
								model.to(__snake_case)

								__lowerCamelCase    :    List[Any] 	=			4
								__lowerCamelCase    :    Union[str, Any] 	=			3
								__lowerCamelCase    :    Optional[Any] 	=			(3_2, 3_2)

								__lowerCamelCase    :    str 	=			torch.ones((batch_size, num_channels) + sizes).to(__snake_case)
								__lowerCamelCase    :    Any 	=			torch.tensor(batch_size * [1E-4]).to(__snake_case)

								with torch.no_grad():
									__lowerCamelCase    :    int 	=			model(__snake_case       ,__snake_case).sample

								__lowerCamelCase    :    Optional[int] 	=			output[0, -3:, -3:, -1].flatten().cpu()
								# fmt: off
								__lowerCamelCase    :    Optional[int] 	=			torch.tensor([-0.0325, -0.0900, -0.0869, -0.0332, -0.0725, -0.0270, -0.0101, 0.0227, 0.0256])
								# fmt: on

								self.assertTrue(torch_all_close(__snake_case       ,__snake_case       ,rtol=1E-2))


							def    lowerCAmelCase					(		self		:      int):
								# not required for this model
								pass







import argparse

import intel_extension_for_pytorch as ipex
import torch

from diffusers import DPMSolverMultistepScheduler, StableDiffusionPipeline


a		=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""")
a		=parser.parse_args()


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

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

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

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

# compute
a		=666
a		=torch.Generator(device).manual_seed(seed)
a		={"""generator""": generator}
if args.steps is not None:
					a		=args.steps

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

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




