

import argparse

import intel_extension_for_pytorch as ipex
import torch

from diffusers import DPMSolverMultistepScheduler, StableDiffusionPipeline


SCREAMING_SNAKE_CASE		:int						=     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""")
SCREAMING_SNAKE_CASE		:Any						=     parser.parse_args()


SCREAMING_SNAKE_CASE		:int						=     """cpu"""
SCREAMING_SNAKE_CASE		:Optional[int]						=     """a lovely <dicoo> in red dress and hat, in the snowly and brightly night, with many brighly buildings"""

SCREAMING_SNAKE_CASE		:Optional[Any]						=     """path-to-your-trained-model"""
SCREAMING_SNAKE_CASE		:str						=     StableDiffusionPipeline.from_pretrained(model_id)
if args.dpm:
 SCREAMING_SNAKE_CASE		:int						=     DPMSolverMultistepScheduler.from_config(pipe.scheduler.config)
SCREAMING_SNAKE_CASE		:List[Any]						=     pipe.to(device)

# to channels last
SCREAMING_SNAKE_CASE		:str						=     pipe.unet.to(memory_format=torch.channels_last)
SCREAMING_SNAKE_CASE		:str						=     pipe.vae.to(memory_format=torch.channels_last)
SCREAMING_SNAKE_CASE		:Union[str, Any]						=     pipe.text_encoder.to(memory_format=torch.channels_last)
if pipe.requires_safety_checker:
 SCREAMING_SNAKE_CASE		:Any						=     pipe.safety_checker.to(memory_format=torch.channels_last)

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

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

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

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



import torch

from diffusers import KDPMaDiscreteScheduler
from diffusers.utils import torch_device

from .test_schedulers import SchedulerCommonTest






class 						__magic_name__					(							snake_case				):
     UpperCamelCase_	:Dict										=      (KDPMaDiscreteScheduler,)
     UpperCamelCase_	:str										=      1_0






     def 	UpperCAmelCase_				(				self		,						**_lowercase	)->     str:
         UpperCamelCase_			  =						{
             "num_train_timesteps": 1_100,
             "beta_start": 0.0_001,
             "beta_end": 0.02,
             "beta_schedule": "linear",
         }

         config.update(**_lowercase	)
         return config






     def 	UpperCAmelCase_				(				self	)->     Union[str, Any]:
         for timesteps in [10, 50, 100, 1_000]:
             self.check_over_configs(num_train_timesteps=_lowercase	)






     def 	UpperCAmelCase_				(				self	)->     int:
         for beta_start, beta_end in zip([0.00_001, 0.0_001, 0.001]		,						[0.0_002, 0.002, 0.02]	):
             self.check_over_configs(beta_start=_lowercase		,						beta_end=_lowercase	)






     def 	UpperCAmelCase_				(				self	)->     str:
         for schedule in ["linear", "scaled_linear"]:
             self.check_over_configs(beta_schedule=_lowercase	)






     def 	UpperCAmelCase_				(				self	)->     Any:
         for prediction_type in ["epsilon", "v_prediction"]:
             self.check_over_configs(prediction_type=_lowercase	)






     def 	UpperCAmelCase_				(				self	)->     Optional[int]:
         UpperCamelCase_			  =						self.scheduler_classes[0]
         UpperCamelCase_			  =						self.get_scheduler_config(prediction_type="v_prediction"	)
         UpperCamelCase_			  =						scheduler_class(**_lowercase	)

         scheduler.set_timesteps(self.num_inference_steps	)

         UpperCamelCase_			  =						self.dummy_model()
         UpperCamelCase_			  =						self.dummy_sample_deter * scheduler.init_noise_sigma
         UpperCamelCase_			  =						sample.to(_lowercase	)

         for i, t in enumerate(scheduler.timesteps	):
             UpperCamelCase_			  =						scheduler.scale_model_input(_lowercase		,						_lowercase	)

             UpperCamelCase_			  =						model(_lowercase		,						_lowercase	)

             UpperCamelCase_			  =						scheduler.step(_lowercase		,						_lowercase		,						_lowercase	)
             UpperCamelCase_			  =						output.prev_sample

         UpperCamelCase_			  =						torch.sum(torch.abs(_lowercase	)	)
         UpperCamelCase_			  =						torch.mean(torch.abs(_lowercase	)	)

         if torch_device in ["cpu", "mps"]:
             assert abs(result_sum.item() - 4.6_9_3_4e-0_7	) < 1e-2
             assert abs(result_mean.item() - 6.1_1_1_2e-1_0	) < 1e-3
         else:
             # CUDA
             assert abs(result_sum.item() - 4.6_9_3_4_2_8_6_5_0_1_7_0_9_7_2e-0_7	) < 1e-2
             assert abs(result_mean.item() - 0.0_002	) < 1e-3






     def 	UpperCAmelCase_				(				self	)->     Dict:
         if torch_device == "mps":
             return
         UpperCamelCase_			  =						self.scheduler_classes[0]
         UpperCamelCase_			  =						self.get_scheduler_config()
         UpperCamelCase_			  =						scheduler_class(**_lowercase	)

         scheduler.set_timesteps(self.num_inference_steps	)

         UpperCamelCase_			  =						self.dummy_model()
         UpperCamelCase_			  =						self.dummy_sample_deter * scheduler.init_noise_sigma
         UpperCamelCase_			  =						sample.to(_lowercase	)

         for i, t in enumerate(scheduler.timesteps	):
             UpperCamelCase_			  =						scheduler.scale_model_input(_lowercase		,						_lowercase	)

             UpperCamelCase_			  =						model(_lowercase		,						_lowercase	)

             UpperCamelCase_			  =						scheduler.step(_lowercase		,						_lowercase		,						_lowercase	)
             UpperCamelCase_			  =						output.prev_sample

         UpperCamelCase_			  =						torch.sum(torch.abs(_lowercase	)	)
         UpperCamelCase_			  =						torch.mean(torch.abs(_lowercase	)	)

         if torch_device in ["cpu", "mps"]:
             assert abs(result_sum.item() - 20.4_125	) < 1e-2
             assert abs(result_mean.item() - 0.0_266	) < 1e-3
         else:
             # CUDA
             assert abs(result_sum.item() - 20.4_125	) < 1e-2
             assert abs(result_mean.item() - 0.0_266	) < 1e-3






     def 	UpperCAmelCase_				(				self	)->     Optional[int]:
         if torch_device == "mps":
             return
         UpperCamelCase_			  =						self.scheduler_classes[0]
         UpperCamelCase_			  =						self.get_scheduler_config()
         UpperCamelCase_			  =						scheduler_class(**_lowercase	)

         scheduler.set_timesteps(self.num_inference_steps		,						device=_lowercase	)

         UpperCamelCase_			  =						self.dummy_model()
         UpperCamelCase_			  =						self.dummy_sample_deter.to(_lowercase	) * scheduler.init_noise_sigma

         for t in scheduler.timesteps:
             UpperCamelCase_			  =						scheduler.scale_model_input(_lowercase		,						_lowercase	)

             UpperCamelCase_			  =						model(_lowercase		,						_lowercase	)

             UpperCamelCase_			  =						scheduler.step(_lowercase		,						_lowercase		,						_lowercase	)
             UpperCamelCase_			  =						output.prev_sample

         UpperCamelCase_			  =						torch.sum(torch.abs(_lowercase	)	)
         UpperCamelCase_			  =						torch.mean(torch.abs(_lowercase	)	)

         if str(_lowercase	).startswith("cpu"	):
             # The following sum varies between 148 and 156 on mps. Why?
             assert abs(result_sum.item() - 20.4_125	) < 1e-2
             assert abs(result_mean.item() - 0.0_266	) < 1e-3
         else:
             # CUDA
             assert abs(result_sum.item() - 20.4_125	) < 1e-2
             assert abs(result_mean.item() - 0.0_266	) < 1e-3
