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


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


_A      :				Union[str, Any] 				=    '''\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     _lowercase					(    lowerCamelCase_      ):


	'''simple docstring'''






	def __init__(				self       :     Optional[Any]				,	SCREAMING_SNAKE_CASE__       :     int				,	SCREAMING_SNAKE_CASE__       :     Any							)  ->   Any:
		super().__init__()

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

		__lowerCAmelCase    							=						None
	def a (				self       :     str				,	SCREAMING_SNAKE_CASE__       :     List[str]=0							)  ->   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"""							)

		__lowerCAmelCase    							=						torch.device(f"""cuda:{gpu_id}"""							)

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

		if self.safety_checker is not None:
			cpu_offload(self.safety_checker				,	execution_device=SCREAMING_SNAKE_CASE__				,	offload_buffers=SCREAMING_SNAKE_CASE__							)
	def a (				self       :     Any				,	SCREAMING_SNAKE_CASE__       :     Optional[Any]=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."""							)

		__lowerCAmelCase    							=						torch.device(f"""cuda:{gpu_id}"""							)

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

		__lowerCAmelCase    							=						None
		for cpu_offloaded_model in [self.unet]:
			__lowerCAmelCase    							=						cpu_offload_with_hook(SCREAMING_SNAKE_CASE__				,	SCREAMING_SNAKE_CASE__				,	prev_module_hook=SCREAMING_SNAKE_CASE__							)

		if self.safety_checker is not None:
			__lowerCAmelCase    							=						cpu_offload_with_hook(self.safety_checker				,	SCREAMING_SNAKE_CASE__				,	prev_module_hook=SCREAMING_SNAKE_CASE__							)

		# We'll offload the last model manually.
		__lowerCAmelCase    							=						hook
	@property
	# Copied from diffusers.pipelines.stable_diffusion.pipeline_stable_diffusion.StableDiffusionPipeline._execution_device
	def a (				self       :     Union[str, Any]							)  ->   Optional[int]:







		if not hasattr(self.unet				,	"""_hf_hook"""							):
			return self.device
		for module in self.unet.modules():
			if (
			    hasattr(SCREAMING_SNAKE_CASE__				,	"""_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       :     int				,	SCREAMING_SNAKE_CASE__       :     str				,	SCREAMING_SNAKE_CASE__       :     Union[str, Any]				,	SCREAMING_SNAKE_CASE__       :     Any				,	SCREAMING_SNAKE_CASE__       :     Optional[int]				,	SCREAMING_SNAKE_CASE__       :     Any				,	SCREAMING_SNAKE_CASE__       :     str				,	SCREAMING_SNAKE_CASE__       :     Optional[int]				,	SCREAMING_SNAKE_CASE__       :     Dict=None							)  ->   Optional[Any]:
		__lowerCAmelCase    							=						(batch_size, num_channels, height, width)
		if isinstance(SCREAMING_SNAKE_CASE__				,	SCREAMING_SNAKE_CASE__							) and len(SCREAMING_SNAKE_CASE__							) != batch_size:
			raise ValueError(
			    f"""You have passed a list of generators of length {len(SCREAMING_SNAKE_CASE__							)}, 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    							=						randn_tensor(SCREAMING_SNAKE_CASE__				,	generator=SCREAMING_SNAKE_CASE__				,	device=SCREAMING_SNAKE_CASE__				,	dtype=SCREAMING_SNAKE_CASE__							)
		else:
			__lowerCAmelCase    							=						latents.to(device=SCREAMING_SNAKE_CASE__				,	dtype=SCREAMING_SNAKE_CASE__							)

		# scale the initial noise by the standard deviation required by the scheduler
		__lowerCAmelCase    							=						latents * self.scheduler.init_noise_sigma
		return latents
	def a (				self       :     Dict				,	SCREAMING_SNAKE_CASE__       :     int				,	SCREAMING_SNAKE_CASE__       :     Optional[Any] = "pil"							)  ->   Any:







		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    							=						(sample / 2 + 0.5).clamp(0				,	1							)
		if output_type == "pt":
			return sample

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

		# Output_type must be 'pil'
		__lowerCAmelCase    							=						self.numpy_to_pil(SCREAMING_SNAKE_CASE__							)
		return sample
	def a (				self       :     Any				,	SCREAMING_SNAKE_CASE__       :     Tuple				,	SCREAMING_SNAKE_CASE__       :     Optional[int]				,	SCREAMING_SNAKE_CASE__       :     List[Any]=None							)  ->   List[str]:







		if self.unet.config.num_class_embeds is not None:
			if isinstance(SCREAMING_SNAKE_CASE__				,	SCREAMING_SNAKE_CASE__							):
				__lowerCAmelCase    							=						torch.tensor(SCREAMING_SNAKE_CASE__				,	dtype=torch.int							)
			elif isinstance(SCREAMING_SNAKE_CASE__				,	SCREAMING_SNAKE_CASE__							):
				assert batch_size == 1, "Batch size must be 1 if classes is an int"
				__lowerCAmelCase    							=						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    							=						torch.randint(0				,	self.unet.config.num_class_embeds				,	size=(batch_size,)							)
			__lowerCAmelCase    							=						class_labels.to(SCREAMING_SNAKE_CASE__							)
		else:
			__lowerCAmelCase    							=						None
		return class_labels
	def a (				self       :     Any				,	SCREAMING_SNAKE_CASE__       :     Any				,	SCREAMING_SNAKE_CASE__       :     Optional[Any]				,	SCREAMING_SNAKE_CASE__       :     List[Any]				,	SCREAMING_SNAKE_CASE__       :     Tuple				,	SCREAMING_SNAKE_CASE__       :     List[str]				,	SCREAMING_SNAKE_CASE__       :     Optional[Any]							)  ->   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:
			__lowerCAmelCase    							=						(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(SCREAMING_SNAKE_CASE__				,	SCREAMING_SNAKE_CASE__							) or callback_steps <= 0)
		):
			raise ValueError(
			    f"""`callback_steps` has to be a positive integer but is {callback_steps} of type"""
			    f""" {type(SCREAMING_SNAKE_CASE__							)}."""							)





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

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

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

		# 3. Handle class_labels for class-conditional models
		__lowerCAmelCase    							=						self.prepare_class_labels(SCREAMING_SNAKE_CASE__				,	SCREAMING_SNAKE_CASE__				,	class_labels=SCREAMING_SNAKE_CASE__							)

		# 4. Prepare timesteps
		if timesteps is not None:
			self.scheduler.set_timesteps(timesteps=SCREAMING_SNAKE_CASE__				,	device=SCREAMING_SNAKE_CASE__							)
			__lowerCAmelCase    							=						self.scheduler.timesteps
			__lowerCAmelCase    							=						len(SCREAMING_SNAKE_CASE__							)
		else:
			self.scheduler.set_timesteps(SCREAMING_SNAKE_CASE__							)
			__lowerCAmelCase    							=						self.scheduler.timesteps

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

				__lowerCAmelCase    							=						self.scheduler.step(SCREAMING_SNAKE_CASE__				,	SCREAMING_SNAKE_CASE__				,	SCREAMING_SNAKE_CASE__				,	generator=SCREAMING_SNAKE_CASE__							)[0]

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

        # 6. Post-process image sample
		__lowerCAmelCase    							=						self.postprocess_image(SCREAMING_SNAKE_CASE__				,	output_type=SCREAMING_SNAKE_CASE__							)

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







'''simple docstring'''

from unittest.mock import patch

import datasets
from datasets import Dataset







def 					UpperCamelCase_						( ) ->					Dict:






 '''simple docstring'''

 __lowerCAmelCase    							=						Dataset.from_dict({"""col_1""": [3, 2, 0, 1]}				)

 with patch("""tqdm.auto.tqdm"""				) as mock_tqdm:
  datasets.disable_progress_bar()
  dset.map(lambda snake_case_				: {"col_2": x["col_1"] + 1}				)
  mock_tqdm.assert_not_called()

  mock_tqdm.reset_mock()

  datasets.enable_progress_bar()
  dset.map(lambda snake_case_				: {"col_2": x["col_1"] + 1}				)
  mock_tqdm.assert_called()






