


'''simple docstring'''


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


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


lowerCamelCase__     						=  '\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 						lowerCAmelCase__      (      lowerCAmelCase__						):




			def __init__(		self	:  Optional[Any]						,	lowerCamelCase__	:  UNetaDModel						,	lowerCamelCase__	:  CMStochasticIterativeScheduler			) ->None:



					'''simple docstring'''







					super().__init__()

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

					_UpperCAmelCase		:							Optional[Any]             =   None




			def 				lowerCAmelCase__     (		self	:  Any						,	lowerCamelCase__	:  Optional[int]=0			) ->int:



					'''simple docstring'''







					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=_UpperCAmelCase			)
							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(_UpperCAmelCase						,	_UpperCAmelCase			)

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




			def 				lowerCAmelCase__     (		self	:  Union[str, Any]						,	lowerCamelCase__	:  Any=0			) ->str:



					'''simple docstring'''







					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		:							Union[str, Any]             =   torch.device(F"""cuda:{gpu_id}"""			)

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

					_UpperCAmelCase		:							List[str]             =   None
					for cpu_offloaded_model in [self.unet]:
							_UpperCAmelCase       ,     _UpperCAmelCase		:							Union[str, Any]             =   cpu_offload_with_hook(_UpperCAmelCase						,	_UpperCAmelCase						,	prev_module_hook=_UpperCAmelCase			)

					if self.safety_checker is not None:
							_UpperCAmelCase       ,     _UpperCAmelCase		:							Union[str, Any]             =   cpu_offload_with_hook(self.safety_checker						,	_UpperCAmelCase						,	prev_module_hook=_UpperCAmelCase			)

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




			@property
			# Copied from diffusers.pipelines.stable_diffusion.pipeline_stable_diffusion.StableDiffusionPipeline._execution_device
			def 				lowerCAmelCase__     (		self	:  Tuple			) ->Any:



					'''simple docstring'''







					if not hasattr(self.unet						,	"_hf_hook"			):
							return self.device
					for module in self.unet.modules():
							if (
							    hasattr(_UpperCAmelCase						,	"_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 				lowerCAmelCase__     (		self	:  str						,	lowerCamelCase__	:  List[Any]						,	lowerCamelCase__	:  Union[str, Any]						,	lowerCamelCase__	:  Union[str, Any]						,	lowerCamelCase__	:  Tuple						,	lowerCamelCase__	:  Any						,	lowerCamelCase__	:  Any						,	lowerCamelCase__	:  List[Any]						,	lowerCamelCase__	:  int=None			) ->List[str]:



					'''simple docstring'''







					_UpperCAmelCase		:							Optional[Any]             =   (batch_size, num_channels, height, width)
					if isinstance(_UpperCAmelCase						,	_UpperCAmelCase			) and len(_UpperCAmelCase			) != batch_size:
							raise ValueError(
							    F"""You have passed a list of generators of length {len(_UpperCAmelCase			)}, 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		:							Any             =   randn_tensor(_UpperCAmelCase						,	generator=_UpperCAmelCase						,	device=_UpperCAmelCase						,	dtype=_UpperCAmelCase			)
					else:
							_UpperCAmelCase		:							Union[str, Any]             =   latents.to(device=_UpperCAmelCase						,	dtype=_UpperCAmelCase			)

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




			def 				lowerCAmelCase__     (		self	:  Optional[int]						,	lowerCamelCase__	:  torch.FloatTensor						,	lowerCamelCase__	:  str = "pil"			) ->str:



					'''simple docstring'''







					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		:							Optional[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		:							List[Any]             =   self.numpy_to_pil(_UpperCAmelCase			)
					return sample




			def 				lowerCAmelCase__     (		self	:  int						,	lowerCamelCase__	:  str						,	lowerCamelCase__	:  Union[str, Any]						,	lowerCamelCase__	:  Optional[Any]=None			) ->str:



					'''simple docstring'''







					if self.unet.config.num_class_embeds is not None:
							if isinstance(_UpperCAmelCase						,	_UpperCAmelCase			):
									_UpperCAmelCase		:							Union[str, Any]             =   torch.tensor(_UpperCAmelCase						,	dtype=torch.int			)
							elif isinstance(_UpperCAmelCase						,	_UpperCAmelCase			):
									assert batch_size == 1, "Batch size must be 1 if classes is an int"
									_UpperCAmelCase		:							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		:							str             =   torch.randint(0						,	self.unet.config.num_class_embeds						,	size=(batch_size,)			)
							_UpperCAmelCase		:							Dict             =   class_labels.to(_UpperCAmelCase			)
					else:
							_UpperCAmelCase		:							List[str]             =   None
					return class_labels




			def 				lowerCAmelCase__     (		self	:  List[str]						,	lowerCamelCase__	:  int						,	lowerCamelCase__	:  Tuple						,	lowerCamelCase__	:  Dict						,	lowerCamelCase__	:  Any						,	lowerCamelCase__	:  Tuple						,	lowerCamelCase__	:  int			) ->Optional[int]:



					'''simple docstring'''







					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		:							str             =   (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(_UpperCAmelCase						,	_UpperCAmelCase			) or callback_steps <= 0)
					):
							raise ValueError(
							    F"""`callback_steps` has to be a positive integer but is {callback_steps} of type"""
							    F""" {type(_UpperCAmelCase			)}."""			)






			@torch.no_grad()
			@replace_example_docstring(_UpperCAmelCase			)
			def __call__(		self	:  Dict						,	lowerCamelCase__	:  int = 1						,	lowerCamelCase__	:  Optional[Union[torch.Tensor, List[int], int]] = None						,	lowerCamelCase__	:  int = 1						,	lowerCamelCase__	:  List[int] = None						,	lowerCamelCase__	:  Optional[Union[torch.Generator, List[torch.Generator]]] = None						,	lowerCamelCase__	:  Optional[torch.FloatTensor] = None						,	lowerCamelCase__	:  Optional[str] = "pil"						,	lowerCamelCase__	:  bool = True						,	lowerCamelCase__	:  Optional[Callable[[int, int, torch.FloatTensor], None]] = None						,	lowerCamelCase__	:  int = 1						,	) ->Tuple:



					'''simple docstring'''







					_UpperCAmelCase		:							Tuple             =   self.unet.config.sample_size
					_UpperCAmelCase		:							List[str]             =   self._execution_device

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

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

					# 3. Handle class_labels for class-conditional models
					_UpperCAmelCase		:							Any             =   self.prepare_class_labels(_UpperCAmelCase						,	_UpperCAmelCase						,	class_labels=_UpperCAmelCase			)

					# 4. Prepare timesteps
					if timesteps is not None:
							self.scheduler.set_timesteps(timesteps=_UpperCAmelCase						,	device=_UpperCAmelCase			)
							_UpperCAmelCase		:							Tuple             =   self.scheduler.timesteps
							_UpperCAmelCase		:							Union[str, Any]             =   len(_UpperCAmelCase			)
					else:
							self.scheduler.set_timesteps(_UpperCAmelCase			)
							_UpperCAmelCase		:							Optional[int]             =   self.scheduler.timesteps

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

									_UpperCAmelCase		:							List[Any]             =   self.scheduler.step(_UpperCAmelCase						,	_UpperCAmelCase						,	_UpperCAmelCase						,	generator=_UpperCAmelCase			)[0]

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

        # 6. Post-process image sample
					_UpperCAmelCase		:							str             =   self.postprocess_image(_UpperCAmelCase						,	output_type=_UpperCAmelCase			)

					# 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=_UpperCAmelCase			)





from ..utils import DummyObject, requires_backends
class       A__  (		metaclass=lowerCAmelCase__							):
							lowerCAmelCase__						:		Tuple            =				["torch", "transformers", "invisible_watermark"]





							def __init__(					self    :  Any	,					*_UpperCAmelCase    :  Optional[Any]	,					**_UpperCAmelCase    :  List[str]							)		->						Tuple:



											"""simple docstring"""


											requires_backends(self	,					['torch', 'transformers', 'invisible_watermark']							)





							@classmethod
							def      a__					(					cls    :  str	,					*_UpperCAmelCase    :  List[str]	,					**_UpperCAmelCase    :  Optional[Any]							)		->						Union[str, Any]:



											"""simple docstring"""


											requires_backends(cls	,					['torch', 'transformers', 'invisible_watermark']							)





							@classmethod
							def      a__					(					cls    :  List[Any]	,					*_UpperCAmelCase    :  List[Any]	,					**_UpperCAmelCase    :  Union[str, Any]							)		->						str:



											"""simple docstring"""


											requires_backends(cls	,					['torch', 'transformers', 'invisible_watermark']							)






class       A__  (		metaclass=lowerCAmelCase__							):
							lowerCAmelCase__						:		Tuple            =				["torch", "transformers", "invisible_watermark"]





							def __init__(					self    :  Dict	,					*_UpperCAmelCase    :  Optional[int]	,					**_UpperCAmelCase    :  Any							)		->						Union[str, Any]:



											"""simple docstring"""


											requires_backends(self	,					['torch', 'transformers', 'invisible_watermark']							)





							@classmethod
							def      a__					(					cls    :  List[str]	,					*_UpperCAmelCase    :  Union[str, Any]	,					**_UpperCAmelCase    :  Optional[Any]							)		->						Optional[int]:



											"""simple docstring"""


											requires_backends(cls	,					['torch', 'transformers', 'invisible_watermark']							)





							@classmethod
							def      a__					(					cls    :  Tuple	,					*_UpperCAmelCase    :  List[Any]	,					**_UpperCAmelCase    :  Tuple							)		->						Any:



											"""simple docstring"""


											requires_backends(cls	,					['torch', 'transformers', 'invisible_watermark']							)





