

"""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


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


__UpperCamelCase						:   List[str]							   =	'''
    Examples:
        ```py
        >>> import torch

        >>> from diffusers import ConsistencyModelPipeline

        >>> device = "cuda"
        >>> # Load the cd_imagenet64_l2 checkpoint.
        >>> model_id_or_path = "openai/diffusers-cd_imagenet64_l2"
        >>> pipe = ConsistencyModelPipeline.from_pretrained(model_id_or_path, torch_dtype=torch.float16)
        >>> pipe.to(device)

        >>> # Onestep Sampling
        >>> image = pipe(num_inference_steps=1).images[0]
        >>> image.save("cd_imagenet64_l2_onestep_sample.png")

        >>> # Onestep sampling, class-conditional image generation
        >>> # ImageNet-64 class label 145 corresponds to king penguins
        >>> image = pipe(num_inference_steps=1, class_labels=145).images[0]
        >>> image.save("cd_imagenet64_l2_onestep_sample_penguin.png")

        >>> # Multistep sampling, class-conditional image generation
        >>> # Timesteps can be explicitly specified; the particular timesteps below are from the original Github repo:
        >>> # https://github.com/openai/consistency_models/blob/main/scripts/launch.sh#L77
        >>> image = pipe(num_inference_steps=None, timesteps=[22, 0], class_labels=145).images[0]
        >>> image.save("cd_imagenet64_l2_multistep_sample_penguin.png")
        ```
'''







class        SCREAMING_SNAKE_CASE      (	a_ ):






					"""simple docstring"""



					def __init__( self    :       Optional[int]     ,lowercase_    :       UNetaDModel     ,lowercase_    :       CMStochasticIterativeScheduler	):
						super().__init__()

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

						lowerCAmelCase__			:		Tuple						       =			None







					def  __lowerCAmelCase				( self    :       Union[str, Any]     ,lowercase_    :       str=0	):
						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'''	)

						lowerCAmelCase__			:		str						       =			torch.device(F'cuda:{gpu_id}'	)

						if self.device.type != "cpu":
							self.to('''cpu'''     ,silence_dtype_warnings=lowercase_	)
							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(lowercase_     ,lowercase_	)

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







					def  __lowerCAmelCase				( self    :       Union[str, Any]     ,lowercase_    :       List[Any]=0	):
						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.'''	)

						lowerCAmelCase__			:		List[Any]						       =			torch.device(F'cuda:{gpu_id}'	)

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

						lowerCAmelCase__			:		Optional[Any]						       =			None
						for cpu_offloaded_model in [self.unet]:
							lowerCAmelCase__       ,lowerCAmelCase__			:		Tuple						       =			cpu_offload_with_hook(lowercase_     ,lowercase_     ,prev_module_hook=lowercase_	)

						if self.safety_checker is not None:
							lowerCAmelCase__       ,lowerCAmelCase__			:		int						       =			cpu_offload_with_hook(self.safety_checker     ,lowercase_     ,prev_module_hook=lowercase_	)

						# We'll offload the last model manually.
						lowerCAmelCase__			:		Dict						       =			hook







					@property
					# Copied from diffusers.pipelines.stable_diffusion.pipeline_stable_diffusion.StableDiffusionPipeline._execution_device
					def  __lowerCAmelCase				( self    :       Union[str, Any]	):
						if not hasattr(self.unet     ,'''_hf_hook'''	):
							return self.device
						for module in self.unet.modules():
							if (
							    hasattr(lowercase_     ,'''_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    :       Any     ,lowercase_    :       Union[str, Any]     ,lowercase_    :       List[str]     ,lowercase_    :       Tuple     ,lowercase_    :       Union[str, Any]     ,lowercase_    :       Tuple     ,lowercase_    :       List[str]     ,lowercase_    :       Optional[int]     ,lowercase_    :       Union[str, Any]=None	):
						lowerCAmelCase__			:		str						       =			(batch_size, num_channels, height, width)
						if isinstance(lowercase_     ,lowercase_	) and len(lowercase_	) != batch_size:
							raise ValueError(
							    F'You have passed a list of generators of length {len(lowercase_	)}, 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:
							lowerCAmelCase__			:		List[Any]						       =			randn_tensor(lowercase_     ,generator=lowercase_     ,device=lowercase_     ,dtype=lowercase_	)
						else:
							lowerCAmelCase__			:		int						       =			latents.to(device=lowercase_     ,dtype=lowercase_	)

						# scale the initial noise by the standard deviation required by the scheduler
						lowerCAmelCase__			:		Tuple						       =			latents * self.scheduler.init_noise_sigma
						return latents







					def  __lowerCAmelCase				( self    :       List[Any]     ,lowercase_    :       torch.FloatTensor     ,lowercase_    :       str = "pil"	):
						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
						lowerCAmelCase__			:		Optional[int]						       =			(sample / 2 + 0.5).clamp(0     ,1	)
						if output_type == "pt":
							return sample

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

						# Output_type must be 'pil'
						lowerCAmelCase__			:		List[Any]						       =			self.numpy_to_pil(lowercase_	)
						return sample







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







					def  __lowerCAmelCase				( self    :       List[Any]     ,lowercase_    :       List[Any]     ,lowercase_    :       Optional[Any]     ,lowercase_    :       Optional[int]     ,lowercase_    :       Optional[int]     ,lowercase_    :       Tuple     ,lowercase_    :       List[Any]	):
						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:
							lowerCAmelCase__			:		List[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(lowercase_     ,lowercase_	) or callback_steps <= 0)
						):
							raise ValueError(
							    F'`callback_steps` has to be a positive integer but is {callback_steps} of type'
							    F' {type(lowercase_	)}.'	)







					@torch.no_grad()
					@replace_example_docstring(lowercase_	)
					def __call__( self    :       int     ,lowercase_    :       int = 1     ,lowercase_    :       Optional[Union[torch.Tensor, List[int], int]] = None     ,lowercase_    :       int = 1     ,lowercase_    :       List[int] = None     ,lowercase_    :       Optional[Union[torch.Generator, List[torch.Generator]]] = None     ,lowercase_    :       Optional[torch.FloatTensor] = None     ,lowercase_    :       Optional[str] = "pil"     ,lowercase_    :       bool = True     ,lowercase_    :       Optional[Callable[[int, int, torch.FloatTensor], None]] = None     ,lowercase_    :       int = 1     ,):
						lowerCAmelCase__			:		List[Any]						       =			self.unet.config.sample_size
						lowerCAmelCase__			:		Dict						       =			self._execution_device

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

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

						# 3. Handle class_labels for class-conditional models
						lowerCAmelCase__			:		str						       =			self.prepare_class_labels(lowercase_     ,lowercase_     ,class_labels=lowercase_	)

						# 4. Prepare timesteps
						if timesteps is not None:
							self.scheduler.set_timesteps(timesteps=lowercase_     ,device=lowercase_	)
							lowerCAmelCase__			:		List[Any]						       =			self.scheduler.timesteps
							lowerCAmelCase__			:		Tuple						       =			len(lowercase_	)
						else:
							self.scheduler.set_timesteps(lowercase_	)
							lowerCAmelCase__			:		Any						       =			self.scheduler.timesteps

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

								lowerCAmelCase__			:		Optional[Any]						       =			self.scheduler.step(lowercase_     ,lowercase_     ,lowercase_     ,generator=lowercase_	)[0]

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

        # 6. Post-process image sample
						lowerCAmelCase__			:		Any						       =			self.postprocess_image(lowercase_     ,output_type=lowercase_	)

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







"""simple docstring"""





def 						__SCREAMING_SNAKE_CASE						(   A_				):
	lowerCAmelCase__			:		List[Any]						       =			53_81
	for x in s:
		lowerCAmelCase__			:		Tuple						       =			((hash_value << 5) + hash_value) + ord(A_				)
	return hash_value & 0Xf_f_f_f_f_f_f_f




