
from typing import Callable, List, Optional, Union

import torch

from ...models import UNetaDModel
from ...schedulers import CMStochasticIterativeScheduler
from ...utils import (
    is_accelerate_available,
    is_accelerate_version,
    logging,
    randn_tensor,
    replace_example_docstring,
)
from ..pipeline_utils import DiffusionPipeline, ImagePipelineOutput


__snake_case					:							int           =							logging.get_logger(__name__)  # pylint: disable=invalid-name


__snake_case					:							List[str]           =							'\n    Examples:\n        ```py\n        >>> import torch\n\n        >>> from diffusers import ConsistencyModelPipeline\n\n        >>> device = "cuda"\n        >>> # Load the cd_imagenet64_l2 checkpoint.\n        >>> model_id_or_path = "openai/diffusers-cd_imagenet64_l2"\n        >>> pipe = ConsistencyModelPipeline.from_pretrained(model_id_or_path, torch_dtype=torch.float16)\n        >>> pipe.to(device)\n\n        >>> # Onestep Sampling\n        >>> image = pipe(num_inference_steps=1).images[0]\n        >>> image.save("cd_imagenet64_l2_onestep_sample.png")\n\n        >>> # Onestep sampling, class-conditional image generation\n        >>> # ImageNet-64 class label 145 corresponds to king penguins\n        >>> image = pipe(num_inference_steps=1, class_labels=145).images[0]\n        >>> image.save("cd_imagenet64_l2_onestep_sample_penguin.png")\n\n        >>> # Multistep sampling, class-conditional image generation\n        >>> # Timesteps can be explicitly specified; the particular timesteps below are from the original Github repo:\n        >>> # https://github.com/openai/consistency_models/blob/main/scripts/launch.sh#L77\n        >>> image = pipe(num_inference_steps=None, timesteps=[22, 0], class_labels=145).images[0]\n        >>> image.save("cd_imagenet64_l2_multistep_sample_penguin.png")\n        ```\n'


class 	UpperCamelCase__       (  UpperCAmelCase__):

						'''simple docstring'''



						def __init__(  self  , A  , A	)  ->None:
									super().__init__()

									self.register_modules(
									    unet=A  , scheduler=A  , )

									UpperCAmelCase__		:Any 		=    None


						def    A__       (  self  , A=0	)  ->Union[str, Any]:
									if is_accelerate_available() and is_accelerate_version('>='  , '0.14.0'	):
												from accelerate import cpu_offload
									else:
												raise ImportError('`enable_sequential_cpu_offload` requires `accelerate v0.14.0` or higher'	)

									UpperCAmelCase__		:Optional[Any] 		=    torch.device(F"""cuda:{gpu_id}"""	)

									if self.device.type != "cpu":
												self.to('cpu'  , silence_dtype_warnings=A	)
												torch.cuda.empty_cache()  # otherwise we don't see the memory savings (but they probably exist)

									for cpu_offloaded_model in [self.unet]:
												cpu_offload(A  , A	)

									if self.safety_checker is not None:
												cpu_offload(self.safety_checker  , execution_device=A  , offload_buffers=A	)


						def    A__       (  self  , A=0	)  ->List[Any]:
									if is_accelerate_available() and is_accelerate_version('>='  , '0.17.0.dev0'	):
												from accelerate import cpu_offload_with_hook
									else:
												raise ImportError('`enable_model_cpu_offload` requires `accelerate v0.17.0` or higher.'	)

									UpperCAmelCase__		:Optional[int] 		=    torch.device(F"""cuda:{gpu_id}"""	)

									if self.device.type != "cpu":
												self.to('cpu'  , silence_dtype_warnings=A	)
												torch.cuda.empty_cache()  # otherwise we don't see the memory savings (but they probably exist)

									UpperCAmelCase__		:Optional[Any] 		=    None
									for cpu_offloaded_model in [self.unet]:
												UpperCAmelCase__		,       UpperCAmelCase__		:List[str] 		=    cpu_offload_with_hook(A  , A  , prev_module_hook=A	)

									if self.safety_checker is not None:
												UpperCAmelCase__		,       UpperCAmelCase__		:Optional[int] 		=    cpu_offload_with_hook(self.safety_checker  , A  , prev_module_hook=A	)

									# We'll offload the last model manually.
									UpperCAmelCase__		:Any 		=    hook


						@property
						# Copied from diffusers.pipelines.stable_diffusion.pipeline_stable_diffusion.StableDiffusionPipeline._execution_device
						def    A__       (  self	)  ->List[Any]:
									if not hasattr(self.unet  , '_hf_hook'	):
												return self.device
									for module in self.unet.modules():
												if (
												    hasattr(A  , '_hf_hook'	)
												    and hasattr(module._hf_hook  , 'execution_device'	)
												    and module._hf_hook.execution_device is not None
												):
															return torch.device(module._hf_hook.execution_device	)
									return self.device


						def    A__       (  self  , A  , A  , A  , A  , A  , A  , A  , A=None	)  ->Dict:
									UpperCAmelCase__		:Union[str, Any] 		=    (batch_size, num_channels, height, width)
									if isinstance(A  , A	) and len(A	) != batch_size:
												raise ValueError(
												    F"""You have passed a list of generators of length {len(A	)}, but requested an effective batch"""
												    F""" size of {batch_size}. Make sure the batch size matches the length of the generators."""	)

									if latents is None:
												UpperCAmelCase__		:List[str] 		=    randn_tensor(A  , generator=A  , device=A  , dtype=A	)
									else:
												UpperCAmelCase__		:Any 		=    latents.to(device=A  , dtype=A	)

									# scale the initial noise by the standard deviation required by the scheduler
									UpperCAmelCase__		:Union[str, Any] 		=    latents * self.scheduler.init_noise_sigma
									return latents


						def    A__       (  self  , A  , A = "pil"	)  ->int:
									if output_type not in ["pt", "np", "pil"]:
												raise ValueError(
												    F"""output_type={output_type} is not supported. Make sure to choose one of ['pt', 'np', or 'pil']"""	)

									# Equivalent to diffusers.VaeImageProcessor.denormalize
									UpperCAmelCase__		:int 		=    (sample / 2 + 0.5).clamp(0  , 1	)
									if output_type == "pt":
												return sample

									# Equivalent to diffusers.VaeImageProcessor.pt_to_numpy
									UpperCAmelCase__		:Union[str, Any] 		=    sample.cpu().permute(0  , 2  , 3  , 1	).numpy()
									if output_type == "np":
												return sample

									# Output_type must be 'pil'
									UpperCAmelCase__		:Tuple 		=    self.numpy_to_pil(A	)
									return sample


						def    A__       (  self  , A  , A  , A=None	)  ->Union[str, Any]:
									if self.unet.config.num_class_embeds is not None:
												if isinstance(A  , A	):
															UpperCAmelCase__		:List[Any] 		=    torch.tensor(A  , dtype=torch.int	)
												elif isinstance(A  , A	):
															assert batch_size == 1, "Batch size must be 1 if classes is an int"
															UpperCAmelCase__		:Optional[Any] 		=    torch.tensor([class_labels]  , dtype=torch.int	)
												elif class_labels is None:
															# Randomly generate batch_size class labels
															# TODO: should use generator here? int analogue of randn_tensor is not exposed in ...utils
															UpperCAmelCase__		:Optional[int] 		=    torch.randint(0  , self.unet.config.num_class_embeds  , size=(batch_size,)	)
												UpperCAmelCase__		:str 		=    class_labels.to(A	)
									else:
												UpperCAmelCase__		:Union[str, Any] 		=    None
									return class_labels


						def    A__       (  self  , A  , A  , A  , A  , A  , A	)  ->List[str]:
									if num_inference_steps is None and timesteps is None:
												raise ValueError('Exactly one of `num_inference_steps` or `timesteps` must be supplied.'	)

									if num_inference_steps is not None and timesteps is not None:
												logger.warning(
												    F"""Both `num_inference_steps`: {num_inference_steps} and `timesteps`: {timesteps} are supplied;"""
												    ' `timesteps` will be used over `num_inference_steps`.'	)

									if latents is not None:
												UpperCAmelCase__		:Union[str, Any] 		=    (batch_size, 3, img_size, img_size)
												if latents.shape != expected_shape:
															raise ValueError(F"""The shape of latents is {latents.shape} but is expected to be {expected_shape}."""	)

									if (callback_steps is None) or (
									    callback_steps is not None and (not isinstance(A  , A	) or callback_steps <= 0)
									):
												raise ValueError(
												    F"""`callback_steps` has to be a positive integer but is {callback_steps} of type"""
												    F""" {type(A	)}."""	)


						@torch.no_grad()
						@replace_example_docstring(A	)
						def __call__(  self  , A = 1  , A = None  , A = 1  , A = None  , A = None  , A = None  , A = "pil"  , A = True  , A = None  , A = 1  , )  ->Tuple:
									UpperCAmelCase__		:Union[str, Any] 		=    self.unet.config.sample_size
									UpperCAmelCase__		:Optional[Any] 		=    self._execution_device

									# 1. Check inputs
									self.check_inputs(A  , A  , A  , A  , A  , A	)

									# 2. Prepare image latents
									# Sample image latents x_0 ~ N(0, sigma_0^2 * I)
									UpperCAmelCase__		:str 		=    self.prepare_latents(
									    batch_size=A  , num_channels=self.unet.config.in_channels  , height=A  , width=A  , dtype=self.unet.dtype  , device=A  , generator=A  , latents=A  , )

									# 3. Handle class_labels for class-conditional models
									UpperCAmelCase__		:Optional[int] 		=    self.prepare_class_labels(A  , A  , class_labels=A	)

									# 4. Prepare timesteps
									if timesteps is not None:
												self.scheduler.set_timesteps(timesteps=A  , device=A	)
												UpperCAmelCase__		:Union[str, Any] 		=    self.scheduler.timesteps
												UpperCAmelCase__		:List[str] 		=    len(A	)
									else:
												self.scheduler.set_timesteps(A	)
												UpperCAmelCase__		:List[str] 		=    self.scheduler.timesteps

									# 5. Denoising loop
									# Multistep sampling: implements Algorithm 1 in the paper
									with self.progress_bar(total=A	) as progress_bar:
												for i, t in enumerate(A	):
															UpperCAmelCase__		:Union[str, Any] 		=    self.scheduler.scale_model_input(A  , A	)
															UpperCAmelCase__		:Optional[int] 		=    self.unet(A  , A  , class_labels=A  , return_dict=A	)[0]

															UpperCAmelCase__		:str 		=    self.scheduler.step(A  , A  , A  , generator=A	)[0]

															# call the callback, if provided
															progress_bar.update()
															if callback is not None and i % callback_steps == 0:
																		callback(A  , A  , A	)

        # 6. Post-process image sample
									UpperCAmelCase__		:List[str] 		=    self.postprocess_image(A  , output_type=A	)

									# Offload last model to CPU
									if hasattr(self  , 'final_offload_hook'	) and self.final_offload_hook is not None:
												self.final_offload_hook.offload()

									if not return_dict:
												return (image,)

									return ImagePipelineOutput(images=A	)







import tempfile

import torch

from diffusers import (
    DEISMultistepScheduler,
    DPMSolverMultistepScheduler,
    DPMSolverSinglestepScheduler,
    UniPCMultistepScheduler,
)

from .test_schedulers import SchedulerCommonTest


class 	UpperCamelCase__       (  UpperCAmelCase__):

						'''simple docstring'''



						__a     :    List[Any]												=			(DPMSolverMultistepScheduler,)
						__a     :    Dict												=			(("""num_inference_steps""", 2_5),)


						def    A__       (  self  , **A	)  ->List[Any]:
									UpperCAmelCase__		:int 		=    {
									    'num_train_timesteps': 10_00,
									    'beta_start': 0.0001,
									    'beta_end': 0.02,
									    'beta_schedule': 'linear',
									    'solver_order': 2,
									    'prediction_type': 'epsilon',
									    'thresholding': False,
									    'sample_max_value': 1.0,
									    'algorithm_type': 'dpmsolver++',
									    'solver_type': 'midpoint',
									    'lower_order_final': False,
									    'lambda_min_clipped': -float('inf'	),
									    'variance_type': None,
									}

									config.update(**A	)
									return config


						def    A__       (  self  , A=0  , **A	)  ->Tuple:
									UpperCAmelCase__		:List[str] 		=    dict(self.forward_default_kwargs	)
									UpperCAmelCase__		:Any 		=    kwargs.pop('num_inference_steps'  , A	)
									UpperCAmelCase__		:int 		=    self.dummy_sample
									UpperCAmelCase__		:Optional[int] 		=    0.1 * sample
									UpperCAmelCase__		:Any 		=    [residual + 0.2, residual + 0.15, residual + 0.10]

									for scheduler_class in self.scheduler_classes:
												UpperCAmelCase__		:Optional[int] 		=    self.get_scheduler_config(**A	)
												UpperCAmelCase__		:List[Any] 		=    scheduler_class(**A	)
												scheduler.set_timesteps(A	)
												# copy over dummy past residuals
												UpperCAmelCase__		:Optional[Any] 		=    dummy_past_residuals[: scheduler.config.solver_order]

												with tempfile.TemporaryDirectory() as tmpdirname:
															scheduler.save_config(A	)
															UpperCAmelCase__		:List[Any] 		=    scheduler_class.from_pretrained(A	)
															new_scheduler.set_timesteps(A	)
															# copy over dummy past residuals
															UpperCAmelCase__		:Union[str, Any] 		=    dummy_past_residuals[: new_scheduler.config.solver_order]

												UpperCAmelCase__		,       UpperCAmelCase__		:int 		=    sample, sample
												for t in range(A  , time_step + scheduler.config.solver_order + 1	):
															UpperCAmelCase__		:Dict 		=    scheduler.step(A  , A  , A  , **A	).prev_sample
															UpperCAmelCase__		:Optional[Any] 		=    new_scheduler.step(A  , A  , A  , **A	).prev_sample

															assert torch.sum(torch.abs(output - new_output	)	) < 1e-5, "Scheduler outputs are not identical"


						def    A__       (  self	)  ->List[Any]:
									pass


						def    A__       (  self  , A=0  , **A	)  ->Tuple:
									UpperCAmelCase__		:List[str] 		=    dict(self.forward_default_kwargs	)
									UpperCAmelCase__		:Dict 		=    kwargs.pop('num_inference_steps'  , A	)
									UpperCAmelCase__		:str 		=    self.dummy_sample
									UpperCAmelCase__		:Union[str, Any] 		=    0.1 * sample
									UpperCAmelCase__		:Union[str, Any] 		=    [residual + 0.2, residual + 0.15, residual + 0.10]

									for scheduler_class in self.scheduler_classes:
												UpperCAmelCase__		:List[Any] 		=    self.get_scheduler_config()
												UpperCAmelCase__		:List[str] 		=    scheduler_class(**A	)
												scheduler.set_timesteps(A	)

												# copy over dummy past residuals (must be after setting timesteps)
												UpperCAmelCase__		:Tuple 		=    dummy_past_residuals[: scheduler.config.solver_order]

												with tempfile.TemporaryDirectory() as tmpdirname:
															scheduler.save_config(A	)
															UpperCAmelCase__		:int 		=    scheduler_class.from_pretrained(A	)
															# copy over dummy past residuals
															new_scheduler.set_timesteps(A	)

															# copy over dummy past residual (must be after setting timesteps)
															UpperCAmelCase__		:Optional[int] 		=    dummy_past_residuals[: new_scheduler.config.solver_order]

												UpperCAmelCase__		:Any 		=    scheduler.step(A  , A  , A  , **A	).prev_sample
												UpperCAmelCase__		:Dict 		=    new_scheduler.step(A  , A  , A  , **A	).prev_sample

												assert torch.sum(torch.abs(output - new_output	)	) < 1e-5, "Scheduler outputs are not identical"


						def    A__       (  self  , A=None  , **A	)  ->List[str]:
									if scheduler is None:
												UpperCAmelCase__		:Optional[Any] 		=    self.scheduler_classes[0]
												UpperCAmelCase__		:Optional[Any] 		=    self.get_scheduler_config(**A	)
												UpperCAmelCase__		:Union[str, Any] 		=    scheduler_class(**A	)

									UpperCAmelCase__		:List[Any] 		=    10
									UpperCAmelCase__		:int 		=    self.dummy_model()
									UpperCAmelCase__		:int 		=    self.dummy_sample_deter
									scheduler.set_timesteps(A	)

									for i, t in enumerate(scheduler.timesteps	):
												UpperCAmelCase__		:Union[str, Any] 		=    model(A  , A	)
												UpperCAmelCase__		:str 		=    scheduler.step(A  , A  , A	).prev_sample

									return sample


						def    A__       (  self	)  ->str:
									UpperCAmelCase__		:Tuple 		=    dict(self.forward_default_kwargs	)

									UpperCAmelCase__		:Optional[Any] 		=    kwargs.pop('num_inference_steps'  , A	)

									for scheduler_class in self.scheduler_classes:
												UpperCAmelCase__		:Union[str, Any] 		=    self.get_scheduler_config()
												UpperCAmelCase__		:Optional[Any] 		=    scheduler_class(**A	)

												UpperCAmelCase__		:Dict 		=    self.dummy_sample
												UpperCAmelCase__		:Dict 		=    0.1 * sample

												if num_inference_steps is not None and hasattr(A  , 'set_timesteps'	):
															scheduler.set_timesteps(A	)
												elif num_inference_steps is not None and not hasattr(A  , 'set_timesteps'	):
															UpperCAmelCase__		:List[Any] 		=    num_inference_steps

												# copy over dummy past residuals (must be done after set_timesteps)
												UpperCAmelCase__		:List[Any] 		=    [residual + 0.2, residual + 0.15, residual + 0.10]
												UpperCAmelCase__		:Any 		=    dummy_past_residuals[: scheduler.config.solver_order]

												UpperCAmelCase__		:str 		=    scheduler.timesteps[5]
												UpperCAmelCase__		:str 		=    scheduler.timesteps[6]

												UpperCAmelCase__		:List[str] 		=    scheduler.step(A  , A  , A  , **A	).prev_sample
												UpperCAmelCase__		:List[Any] 		=    scheduler.step(A  , A  , A  , **A	).prev_sample

												self.assertEqual(output_a.shape  , sample.shape	)
												self.assertEqual(output_a.shape  , output_a.shape	)


						def    A__       (  self	)  ->Optional[Any]:
									for timesteps in [25, 50, 1_00, 9_99, 10_00]:
												self.check_over_configs(num_train_timesteps=A	)


						def    A__       (  self	)  ->str:
									self.check_over_configs(thresholding=A	)
									for order in [1, 2, 3]:
												for solver_type in ["midpoint", "heun"]:
															for threshold in [0.5, 1.0, 2.0]:
																		for prediction_type in ["epsilon", "sample"]:
																					self.check_over_configs(
																					    thresholding=A  , prediction_type=A  , sample_max_value=A  , algorithm_type='dpmsolver++'  , solver_order=A  , solver_type=A  , )


						def    A__       (  self	)  ->Optional[int]:
									for prediction_type in ["epsilon", "v_prediction"]:
												self.check_over_configs(prediction_type=A	)


						def    A__       (  self	)  ->List[str]:
									for algorithm_type in ["dpmsolver", "dpmsolver++", "sde-dpmsolver", "sde-dpmsolver++"]:
												for solver_type in ["midpoint", "heun"]:
															for order in [1, 2, 3]:
																		for prediction_type in ["epsilon", "sample"]:
																					if algorithm_type in ["sde-dpmsolver", "sde-dpmsolver++"]:
																								if order == 3:
																											continue
																					else:
																								self.check_over_configs(
																								    solver_order=A  , solver_type=A  , prediction_type=A  , algorithm_type=A  , )
																					UpperCAmelCase__		:Optional[int] 		=    self.full_loop(
																					    solver_order=A  , solver_type=A  , prediction_type=A  , algorithm_type=A  , )
																					assert not torch.isnan(A	).any(), "Samples have nan numbers"


						def    A__       (  self	)  ->List[Any]:
									self.check_over_configs(lower_order_final=A	)
									self.check_over_configs(lower_order_final=A	)


						def    A__       (  self	)  ->Optional[Any]:
									self.check_over_configs(lambda_min_clipped=-float('inf'	)	)
									self.check_over_configs(lambda_min_clipped=-5.1	)


						def    A__       (  self	)  ->Union[str, Any]:
									self.check_over_configs(variance_type=A	)
									self.check_over_configs(variance_type='learned_range'	)


						def    A__       (  self	)  ->List[str]:
									for num_inference_steps in [1, 2, 3, 5, 10, 50, 1_00, 9_99, 10_00]:
												self.check_over_forward(num_inference_steps=A  , time_step=0	)


						def    A__       (  self	)  ->int:
									UpperCAmelCase__		:int 		=    self.full_loop()
									UpperCAmelCase__		:Optional[int] 		=    torch.mean(torch.abs(A	)	)

									assert abs(result_mean.item() - 0.3301	) < 1e-3


						def    A__       (  self	)  ->Any:
									UpperCAmelCase__		:Optional[int] 		=    self.full_loop(thresholding=A  , dynamic_thresholding_ratio=0.87  , sample_max_value=0.5	)
									UpperCAmelCase__		:Any 		=    torch.mean(torch.abs(A	)	)

									assert abs(result_mean.item() - 1.1364	) < 1e-3


						def    A__       (  self	)  ->Union[str, Any]:
									UpperCAmelCase__		:Optional[Any] 		=    self.full_loop(prediction_type='v_prediction'	)
									UpperCAmelCase__		:int 		=    torch.mean(torch.abs(A	)	)

									assert abs(result_mean.item() - 0.2251	) < 1e-3


						def    A__       (  self	)  ->str:
									UpperCAmelCase__		:Optional[int] 		=    self.full_loop(prediction_type='v_prediction'  , use_karras_sigmas=A	)
									UpperCAmelCase__		:Optional[int] 		=    torch.mean(torch.abs(A	)	)

									assert abs(result_mean.item() - 0.2096	) < 1e-3


						def    A__       (  self	)  ->List[Any]:
									# make sure that iterating over schedulers with same config names gives same results
									# for defaults
									UpperCAmelCase__		:Tuple 		=    DPMSolverMultistepScheduler(**self.get_scheduler_config()	)
									UpperCAmelCase__		:Optional[int] 		=    self.full_loop(scheduler=A	)
									UpperCAmelCase__		:List[Any] 		=    torch.mean(torch.abs(A	)	)

									assert abs(result_mean.item() - 0.3301	) < 1e-3

									UpperCAmelCase__		:Dict 		=    DPMSolverSinglestepScheduler.from_config(scheduler.config	)
									UpperCAmelCase__		:List[str] 		=    UniPCMultistepScheduler.from_config(scheduler.config	)
									UpperCAmelCase__		:Union[str, Any] 		=    DEISMultistepScheduler.from_config(scheduler.config	)
									UpperCAmelCase__		:List[str] 		=    DPMSolverMultistepScheduler.from_config(scheduler.config	)

									UpperCAmelCase__		:Dict 		=    self.full_loop(scheduler=A	)
									UpperCAmelCase__		:Dict 		=    torch.mean(torch.abs(A	)	)

									assert abs(result_mean.item() - 0.3301	) < 1e-3


						def    A__       (  self	)  ->Dict:
									UpperCAmelCase__		:Tuple 		=    self.scheduler_classes[0]
									UpperCAmelCase__		:List[str] 		=    self.get_scheduler_config(thresholding=A  , dynamic_thresholding_ratio=0	)
									UpperCAmelCase__		:str 		=    scheduler_class(**A	)

									UpperCAmelCase__		:int 		=    10
									UpperCAmelCase__		:Optional[int] 		=    self.dummy_model()
									UpperCAmelCase__		:List[str] 		=    self.dummy_sample_deter.half()
									scheduler.set_timesteps(A	)

									for i, t in enumerate(scheduler.timesteps	):
												UpperCAmelCase__		:str 		=    model(A  , A	)
												UpperCAmelCase__		:Union[str, Any] 		=    scheduler.step(A  , A  , A	).prev_sample

									assert sample.dtype == torch.floataa


						def    A__       (  self  , **A	)  ->Any:
									for scheduler_class in self.scheduler_classes:
												UpperCAmelCase__		:Optional[int] 		=    self.get_scheduler_config(**A	)
												UpperCAmelCase__		:int 		=    scheduler_class(**A	)

												scheduler.set_timesteps(scheduler.config.num_train_timesteps	)
												assert len(scheduler.timesteps.unique()	) == scheduler.num_inference_steps





