"""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_:       Optional[int]			     =       logging.get_logger(__name__)  # pylint: disable=invalid-name


lowerCAmelCase_:       Dict			     =       "\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  a__     ( lowerCamelCase_						):


   def __init__(  self,					_UpperCAmelCase,					_UpperCAmelCase	):
        '''simple docstring'''



        super().__init__()

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

        lowercase__								=				None


   def 						snake_case__						(  self,					_UpperCAmelCase=0	):
        '''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"	)

        lowercase__								=				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 						snake_case__						(  self,					_UpperCAmelCase=0	):
        '''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."	)

        lowercase__								=				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)

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

        if self.safety_checker is not None:
             lowercase__								=				cpu_offload_with_hook(self.safety_checker,					__SCREAMING_SNAKE_CASE,					prev_module_hook=__SCREAMING_SNAKE_CASE	)

        # We'll offload the last model manually.
        lowercase__								=				hook


   @property
   # Copied from diffusers.pipelines.stable_diffusion.pipeline_stable_diffusion.StableDiffusionPipeline._execution_device
   def 						snake_case__						(  self	):
        '''simple docstring'''



        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 						snake_case__						(  self,					_UpperCAmelCase,					_UpperCAmelCase,					_UpperCAmelCase,					_UpperCAmelCase,					_UpperCAmelCase,					_UpperCAmelCase,					_UpperCAmelCase,					_UpperCAmelCase=None	):
        '''simple docstring'''



        lowercase__								=				(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:
             lowercase__								=				randn_tensor(__SCREAMING_SNAKE_CASE,					generator=__SCREAMING_SNAKE_CASE,					device=__SCREAMING_SNAKE_CASE,					dtype=__SCREAMING_SNAKE_CASE	)
        else:
             lowercase__								=				latents.to(device=__SCREAMING_SNAKE_CASE,					dtype=__SCREAMING_SNAKE_CASE	)

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


   def 						snake_case__						(  self,					_UpperCAmelCase,					_UpperCAmelCase = "pil"	):
        '''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
        lowercase__								=				(sample / 2 + 0.5).clamp(0,					1	)
        if output_type == "pt":
             return sample

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

        # Output_type must be 'pil'
        lowercase__								=				self.numpy_to_pil(__SCREAMING_SNAKE_CASE	)
        return sample


   def 						snake_case__						(  self,					_UpperCAmelCase,					_UpperCAmelCase,					_UpperCAmelCase=None	):
        '''simple docstring'''



        if self.unet.config.num_class_embeds is not None:
             if isinstance(__SCREAMING_SNAKE_CASE,					__SCREAMING_SNAKE_CASE	):
                  lowercase__								=				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"
                  lowercase__								=				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
                  lowercase__								=				torch.randint(0,					self.unet.config.num_class_embeds,					size=(batch_size,)	)
             lowercase__								=				class_labels.to(__SCREAMING_SNAKE_CASE	)
        else:
             lowercase__								=				None
        return class_labels


   def 						snake_case__						(  self,					_UpperCAmelCase,					_UpperCAmelCase,					_UpperCAmelCase,					_UpperCAmelCase,					_UpperCAmelCase,					_UpperCAmelCase	):
        '''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:
             lowercase__								=				(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,					_UpperCAmelCase = 1,					_UpperCAmelCase = None,					_UpperCAmelCase = 1,					_UpperCAmelCase = None,					_UpperCAmelCase = None,					_UpperCAmelCase = None,					_UpperCAmelCase = "pil",					_UpperCAmelCase = True,					_UpperCAmelCase = None,					_UpperCAmelCase = 1,					):
        '''simple docstring'''



        lowercase__								=				self.unet.config.sample_size
        lowercase__								=				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)
        lowercase__								=				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
        lowercase__								=				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	)
             lowercase__								=				self.scheduler.timesteps
             lowercase__								=				len(__SCREAMING_SNAKE_CASE	)
        else:
             self.scheduler.set_timesteps(__SCREAMING_SNAKE_CASE	)
             lowercase__								=				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	):
                  lowercase__								=				self.scheduler.scale_model_input(__SCREAMING_SNAKE_CASE,					__SCREAMING_SNAKE_CASE	)
                  lowercase__								=				self.unet(__SCREAMING_SNAKE_CASE,					__SCREAMING_SNAKE_CASE,					class_labels=__SCREAMING_SNAKE_CASE,					return_dict=__SCREAMING_SNAKE_CASE	)[0]

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







import posixpath
from pathlib import Path
from unittest.mock import patch

import pytest
from fsspec.implementations.local import AbstractFileSystem, LocalFileSystem, stringify_path
from fsspec.registry import _registry as _fsspec_registry






class 					lowerCAmelCase__  (    lowerCamelCase_	):
					lowerCAmelCase_	  =	'''mock'''






					def __init__(  self							,				*__SCREAMING_SNAKE_CASE							,				__SCREAMING_SNAKE_CASE							,				**__SCREAMING_SNAKE_CASE						):


											"""simple docstring"""





											super().__init__()
											lowercase_      :	int						=			LocalFileSystem(*__SCREAMING_SNAKE_CASE							,				**__SCREAMING_SNAKE_CASE						)
											lowercase_      :	Tuple						=			Path(__SCREAMING_SNAKE_CASE						).resolve().as_posix() + '''/'''






					def       _snake_case       (  self							,				__SCREAMING_SNAKE_CASE							,				*__SCREAMING_SNAKE_CASE							,				**__SCREAMING_SNAKE_CASE						):


											"""simple docstring"""





											lowercase_      :	List[str]						=			posixpath.join(self.local_root_dir							,				self._strip_protocol(__SCREAMING_SNAKE_CASE						)						)
											return self._fs.mkdir(__SCREAMING_SNAKE_CASE							,				*__SCREAMING_SNAKE_CASE							,				**__SCREAMING_SNAKE_CASE						)






					def       _snake_case       (  self							,				__SCREAMING_SNAKE_CASE							,				*__SCREAMING_SNAKE_CASE							,				**__SCREAMING_SNAKE_CASE						):


											"""simple docstring"""





											lowercase_      :	Optional[int]						=			posixpath.join(self.local_root_dir							,				self._strip_protocol(__SCREAMING_SNAKE_CASE						)						)
											return self._fs.makedirs(__SCREAMING_SNAKE_CASE							,				*__SCREAMING_SNAKE_CASE							,				**__SCREAMING_SNAKE_CASE						)






					def       _snake_case       (  self							,				__SCREAMING_SNAKE_CASE						):


											"""simple docstring"""





											lowercase_      :	Any						=			posixpath.join(self.local_root_dir							,				self._strip_protocol(__SCREAMING_SNAKE_CASE						)						)
											return self._fs.rmdir(__SCREAMING_SNAKE_CASE						)






					def       _snake_case       (  self							,				__SCREAMING_SNAKE_CASE							,				__SCREAMING_SNAKE_CASE=True							,				*__SCREAMING_SNAKE_CASE							,				**__SCREAMING_SNAKE_CASE						):


											"""simple docstring"""





											lowercase_      :	str						=			posixpath.join(self.local_root_dir							,				self._strip_protocol(__SCREAMING_SNAKE_CASE						)						)
											lowercase_      :	Union[str, Any]						=			self._fs.ls(__SCREAMING_SNAKE_CASE							,				detail=__SCREAMING_SNAKE_CASE							,				*__SCREAMING_SNAKE_CASE							,				**__SCREAMING_SNAKE_CASE						)
											if detail:
																	return [{**info, "name": info["name"][len(self.local_root_dir						) :]} for info in out]
											else:
																	return [name[len(self.local_root_dir						) :] for name in out]






					def       _snake_case       (  self							,				__SCREAMING_SNAKE_CASE							,				*__SCREAMING_SNAKE_CASE							,				**__SCREAMING_SNAKE_CASE						):


											"""simple docstring"""





											lowercase_      :	Optional[int]						=			posixpath.join(self.local_root_dir							,				self._strip_protocol(__SCREAMING_SNAKE_CASE						)						)
											lowercase_      :	Optional[Any]						=			dict(self._fs.info(__SCREAMING_SNAKE_CASE							,				*__SCREAMING_SNAKE_CASE							,				**__SCREAMING_SNAKE_CASE						)						)
											lowercase_      :	Dict						=			out['''name'''][len(self.local_root_dir						) :]
											return out






					def       _snake_case       (  self							,				__SCREAMING_SNAKE_CASE							,				__SCREAMING_SNAKE_CASE							,				*__SCREAMING_SNAKE_CASE							,				**__SCREAMING_SNAKE_CASE						):


											"""simple docstring"""





											lowercase_      :	List[str]						=			posixpath.join(self.local_root_dir							,				self._strip_protocol(__SCREAMING_SNAKE_CASE						)						)
											lowercase_      :	str						=			posixpath.join(self.local_root_dir							,				self._strip_protocol(__SCREAMING_SNAKE_CASE						)						)
											return self._fs.cp_file(__SCREAMING_SNAKE_CASE							,				__SCREAMING_SNAKE_CASE							,				*__SCREAMING_SNAKE_CASE							,				**__SCREAMING_SNAKE_CASE						)






					def       _snake_case       (  self							,				__SCREAMING_SNAKE_CASE							,				*__SCREAMING_SNAKE_CASE							,				**__SCREAMING_SNAKE_CASE						):


											"""simple docstring"""





											lowercase_      :	List[Any]						=			posixpath.join(self.local_root_dir							,				self._strip_protocol(__SCREAMING_SNAKE_CASE						)						)
											return self._fs.rm_file(__SCREAMING_SNAKE_CASE							,				*__SCREAMING_SNAKE_CASE							,				**__SCREAMING_SNAKE_CASE						)






					def       _snake_case       (  self							,				__SCREAMING_SNAKE_CASE							,				*__SCREAMING_SNAKE_CASE							,				**__SCREAMING_SNAKE_CASE						):


											"""simple docstring"""





											lowercase_      :	Union[str, Any]						=			posixpath.join(self.local_root_dir							,				self._strip_protocol(__SCREAMING_SNAKE_CASE						)						)
											return self._fs.rm(__SCREAMING_SNAKE_CASE							,				*__SCREAMING_SNAKE_CASE							,				**__SCREAMING_SNAKE_CASE						)






					def       _snake_case       (  self							,				__SCREAMING_SNAKE_CASE							,				*__SCREAMING_SNAKE_CASE							,				**__SCREAMING_SNAKE_CASE						):


											"""simple docstring"""





											lowercase_      :	Union[str, Any]						=			posixpath.join(self.local_root_dir							,				self._strip_protocol(__SCREAMING_SNAKE_CASE						)						)
											return self._fs._open(__SCREAMING_SNAKE_CASE							,				*__SCREAMING_SNAKE_CASE							,				**__SCREAMING_SNAKE_CASE						)






					def       _snake_case       (  self							,				__SCREAMING_SNAKE_CASE						):


											"""simple docstring"""





											lowercase_      :	Dict						=			posixpath.join(self.local_root_dir							,				self._strip_protocol(__SCREAMING_SNAKE_CASE						)						)
											return self._fs.created(__SCREAMING_SNAKE_CASE						)






					def       _snake_case       (  self							,				__SCREAMING_SNAKE_CASE						):


											"""simple docstring"""





											lowercase_      :	Optional[Any]						=			posixpath.join(self.local_root_dir							,				self._strip_protocol(__SCREAMING_SNAKE_CASE						)						)
											return self._fs.modified(__SCREAMING_SNAKE_CASE						)






					@classmethod
					def       _snake_case       (  cls							,				__SCREAMING_SNAKE_CASE						):


											"""simple docstring"""





											lowercase_      :	Optional[Any]						=			stringify_path(__SCREAMING_SNAKE_CASE						)
											if path.startswith('''mock://'''						):
																	lowercase_      :	int						=			path[7:]
											return path







class 					lowerCAmelCase__  (    lowerCamelCase_	):
					lowerCAmelCase_	  =	'''tmp'''
					lowerCAmelCase_	  =	None






					def __init__(  self							,				*__SCREAMING_SNAKE_CASE							,				**__SCREAMING_SNAKE_CASE						):


											"""simple docstring"""





											assert self.tmp_dir is not None, "TmpDirFileSystem.tmp_dir is not set"
											super().__init__(*__SCREAMING_SNAKE_CASE							,				**__SCREAMING_SNAKE_CASE							,				local_root_dir=self.tmp_dir							,				auto_mkdir=__SCREAMING_SNAKE_CASE						)






					@classmethod
					def       _snake_case       (  cls							,				__SCREAMING_SNAKE_CASE						):


											"""simple docstring"""





											lowercase_      :	str						=			stringify_path(__SCREAMING_SNAKE_CASE						)
											if path.startswith('''tmp://'''						):
																	lowercase_      :	Union[str, Any]						=			path[6:]
											return path







@pytest.fixture
def    snake_case_						(							):



						"""simple docstring"""





						lowercase_      :	List[Any]						=			MockFileSystem
						lowercase_      :	Tuple						=			TmpDirFileSystem
						yield
						del _fsspec_registry["mock"]
						del _fsspec_registry["tmp"]
@pytest.fixture
def    snake_case_						(							__SCREAMING_SNAKE_CASE   :					List[Any] , __SCREAMING_SNAKE_CASE   :					List[Any]				):



						"""simple docstring"""





						lowercase_      :	List[str]						=			tmp_path_factory.mktemp('''mockfs'''				)
						return MockFileSystem(local_root_dir=__SCREAMING_SNAKE_CASE , auto_mkdir=__SCREAMING_SNAKE_CASE				)




@pytest.fixture
def    snake_case_						(							__SCREAMING_SNAKE_CASE   :					Union[str, Any] , __SCREAMING_SNAKE_CASE   :					Any				):



						"""simple docstring"""





						lowercase_      :	Union[str, Any]						=			tmp_path_factory.mktemp('''tmpfs'''				)
						with patch.object(__SCREAMING_SNAKE_CASE , '''tmp_dir''' , __SCREAMING_SNAKE_CASE				):
												yield TmpDirFileSystem()



