

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


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


lowercase_									=    "\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		(				_UpperCAmelCase					):







    """simple docstring"""





    def __init__( self					:			str,lowercase_					:			UNetaDModel,lowercase_					:			CMStochasticIterativeScheduler						)->       None:





        '''simple docstring'''







        super().__init__()

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

        A__        =  None


    def 		snake_case__	( self					:			int,lowercase_					:			Any=0						)->       Tuple:





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

        A__        =  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 		snake_case__	( self					:			Tuple,lowercase_					:			int=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.'						)

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

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

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

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


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





        '''simple docstring'''







        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 		snake_case__	( self					:			int,lowercase_					:			str,lowercase_					:			int,lowercase_					:			Optional[Any],lowercase_					:			Optional[Any],lowercase_					:			Dict,lowercase_					:			List[Any],lowercase_					:			Dict,lowercase_					:			Optional[Any]=None						)->       List[Any]:





        '''simple docstring'''







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

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


    def 		snake_case__	( self					:			str,lowercase_					:			torch.FloatTensor,lowercase_					:			str = "pil"						)->       List[Any]:





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

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

        # Output_type must be 'pil'
        A__        =  self.numpy_to_pil(lowercase_						)
        return sample


    def 		snake_case__	( self					:			List[str],lowercase_					:			List[str],lowercase_					:			Tuple,lowercase_					:			Any=None						)->       Optional[Any]:





        '''simple docstring'''







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


    def 		snake_case__	( self					:			Dict,lowercase_					:			str,lowercase_					:			Tuple,lowercase_					:			List[str],lowercase_					:			int,lowercase_					:			Optional[Any],lowercase_					:			Any						)->       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:
            A__        =  (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					:			Tuple,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,)->       Any:





        '''simple docstring'''







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

                A__        =  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
        A__        =  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_						)




import enum
import inspect
import warnings
from functools import wraps
from typing import Callable, Optional

from .logging import get_logger


lowercase_									=    set()
lowercase_									=    get_logger(__name__)
def 		_snake_case(       SCREAMING_SNAKE_CASE__     :      Optional[str] = None )				-> Optional[int]:





    '''simple docstring'''






    def decorator(SCREAMING_SNAKE_CASE__     :      Callable ):
        global _emitted_deprecation_warnings

        if inspect.isclass(SCREAMING_SNAKE_CASE__ ):
            A__        =  deprecated_class_or_function.__init__
            A__        =  deprecated_class_or_function.__name__
        else:
            A__        =  deprecated_class_or_function
            A__        =  deprecated_function.__name__
            # Support deprecating __init__ class method: class name instead
            A__        =  name if name != '__init__' else deprecated_function.__qualname__.split('.' )[-2]

        A__        =  (
            f'{name} is deprecated and will be removed in the next major version of datasets.' + f' {help_message}'
            if help_message
            else ''
        )

        @wraps(SCREAMING_SNAKE_CASE__ )
        def wrapper(*SCREAMING_SNAKE_CASE__     :      Tuple       ,     **SCREAMING_SNAKE_CASE__     :      Any ):
            A__        =  hash(SCREAMING_SNAKE_CASE__ )
            if func_hash not in _emitted_deprecation_warnings:
                warnings.warn(SCREAMING_SNAKE_CASE__       ,     category=SCREAMING_SNAKE_CASE__       ,     stacklevel=2 )
                _emitted_deprecation_warnings.add(SCREAMING_SNAKE_CASE__ )
            return deprecated_function(*SCREAMING_SNAKE_CASE__       ,     **SCREAMING_SNAKE_CASE__ )

        A__        =  'deprecated'

        if inspect.isclass(SCREAMING_SNAKE_CASE__ ):
            A__        =  wrapper
            return deprecated_class_or_function
        else:
            return wrapper

    return decorator


class 					A		(				enum.EnumMeta					):







    """simple docstring"""





    def __getattribute__( cls					:			Optional[Any],lowercase_					:			int						)->       Optional[Any]:





        '''simple docstring'''







        A__        =  super().__getattribute__(lowercase_						)
        if isinstance(lowercase_,enum.Enum						) and obj._on_access:
            obj._on_access()
        return obj


    def __getitem__( cls					:			int,lowercase_					:			Any						)->       Tuple:





        '''simple docstring'''







        A__        =  super().__getitem__(lowercase_						)
        if member._on_access:
            member._on_access()
        return member


    def __call__( cls					:			Optional[int],lowercase_					:			Tuple,lowercase_					:			List[Any]=None,*, lowercase_					:			Any=None,lowercase_					:			Optional[Any]=None,lowercase_					:			Any=None,lowercase_					:			Optional[int]=1						)->       List[str]:





        '''simple docstring'''







        A__        =  super().__call__(lowercase_,lowercase_,module=lowercase_,qualname=lowercase_,type=lowercase_,start=lowercase_						)
        if isinstance(lowercase_,enum.Enum						) and obj._on_access:
            obj._on_access()
        return obj




class 					A		(				enum.Enum       ,				metaclass=_UpperCAmelCase					):







    """simple docstring"""





    def __new__( cls					:			Any,lowercase_					:			str						)->       Any:





        '''simple docstring'''







        A__        =  object.__new__(cls						)
        A__        =  value
        A__        =  member.deprecate
        return member


    @property
    def 		snake_case__	( self					:			List[Any]						)->       List[Any]:





        '''simple docstring'''







        return ""


    def 		snake_case__	( self					:			List[str]						)->       Dict:





        '''simple docstring'''







        A__        =  F' {self.help_message}' if self.help_message else ''
        warnings.warn(
            F'\'{self.__objclass__.__name__}\' is deprecated and will be removed in the next major version of datasets.'
            + help_message,lowercase_,stacklevel=3,)

