





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


UpperCAmelCase_       : List[Any]							=    """
    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   lowercase__     (				__A	):







  def __init__(						self     ,    _lowercase     ,    _lowercase    ):
       super().__init__()

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

       lowerCAmelCase_       :							Optional[int]			 =		None







  def 						UpperCAmelCase__		(						self     ,    _lowercase=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_       :							Optional[int]			 =		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 						UpperCAmelCase__		(						self     ,    _lowercase=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_       :							Dict			 =		None
       for cpu_offloaded_model in [self.unet]:
            lowerCAmelCase_ ,	lowerCAmelCase_       :							Optional[Any]			 =		cpu_offload_with_hook(_lowercase     ,    _lowercase     ,    prev_module_hook=_lowercase    )

       if self.safety_checker is not None:
            lowerCAmelCase_ ,	lowerCAmelCase_       :							Tuple			 =		cpu_offload_with_hook(self.safety_checker     ,    _lowercase     ,    prev_module_hook=_lowercase    )

       # We'll offload the last model manually.
       lowerCAmelCase_       :							List[str]			 =		hook







  @property
  # Copied from diffusers.pipelines.stable_diffusion.pipeline_stable_diffusion.StableDiffusionPipeline._execution_device
  def 						UpperCAmelCase__		(						self    ):
       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 						UpperCAmelCase__		(						self     ,    _lowercase     ,    _lowercase     ,    _lowercase     ,    _lowercase     ,    _lowercase     ,    _lowercase     ,    _lowercase     ,    _lowercase=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_       :							Union[str, Any]			 =		latents * self.scheduler.init_noise_sigma
       return latents







  def 						UpperCAmelCase__		(						self     ,    _lowercase     ,    _lowercase = "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_       :							Any			 =		(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_       :							Optional[Any]			 =		self.numpy_to_pil(_lowercase    )
       return sample







  def 						UpperCAmelCase__		(						self     ,    _lowercase     ,    _lowercase     ,    _lowercase=None    ):
       if self.unet.config.num_class_embeds is not None:
            if isinstance(_lowercase     ,    _lowercase    ):
                 lowerCAmelCase_       :							str			 =		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_       :							Dict			 =		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_       :							Union[str, Any]			 =		torch.randint(0     ,    self.unet.config.num_class_embeds     ,    size=(batch_size,)    )
            lowerCAmelCase_       :							List[str]			 =		class_labels.to(_lowercase    )
       else:
            lowerCAmelCase_       :							Optional[Any]			 =		None
       return class_labels







  def 						UpperCAmelCase__		(						self     ,    _lowercase     ,    _lowercase     ,    _lowercase     ,    _lowercase     ,    _lowercase     ,    _lowercase    ):
       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_       :							Tuple			 =		(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     ,    _lowercase = 1     ,    _lowercase = None     ,    _lowercase = 1     ,    _lowercase = None     ,    _lowercase = None     ,    _lowercase = None     ,    _lowercase = "pil"     ,    _lowercase = True     ,    _lowercase = None     ,    _lowercase = 1     ,    ):
       lowerCAmelCase_       :							Tuple			 =		self.unet.config.sample_size
       lowerCAmelCase_       :							List[str]			 =		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_       :							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_       :							List[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_       :							int			 =		self.scheduler.timesteps
            lowerCAmelCase_       :							List[Any]			 =		len(_lowercase    )
       else:
            self.scheduler.set_timesteps(_lowercase    )
            lowerCAmelCase_       :							List[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_       :							str			 =		self.scheduler.scale_model_input(_lowercase     ,    _lowercase    )
                 lowerCAmelCase_       :							str			 =		self.unet(_lowercase     ,    _lowercase     ,    class_labels=_lowercase     ,    return_dict=_lowercase    )[0]

                 lowerCAmelCase_       :							List[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    )







import copy

from ...configuration_utils import PretrainedConfig
from ...utils import logging
from ..auto.configuration_auto import AutoConfig
from ..clip.configuration_clip import CLIPVisionConfig


UpperCAmelCase_       : Union[str, Any]							=    logging.get_logger(__name__)





class   lowercase__     (				__A	):
  __UpperCamelCase						  =	"""vision-text-dual-encoder"""
  __UpperCamelCase						  =	True







  def __init__(						self     ,    _lowercase=512     ,    _lowercase=2.6592     ,    **_lowercase    ):
       super().__init__(**_lowercase    )

       if "vision_config" not in kwargs:
            raise ValueError("""`vision_config` can not be `None`."""    )

       if "text_config" not in kwargs:
            raise ValueError("""`text_config` can not be `None`."""    )

       lowerCAmelCase_       :							Dict			 =		kwargs.pop("""vision_config"""    )
       lowerCAmelCase_       :							Optional[int]			 =		kwargs.pop("""text_config"""    )

       lowerCAmelCase_       :							Tuple			 =		vision_config.pop("""model_type"""    )
       lowerCAmelCase_       :							str			 =		text_config.pop("""model_type"""    )

       if vision_model_type == "clip":
            lowerCAmelCase_       :							Dict			 =		AutoConfig.for_model(_lowercase     ,    **_lowercase    ).vision_config
       elif vision_model_type == "clip_vision_model":
            lowerCAmelCase_       :							Any			 =		CLIPVisionConfig(**_lowercase    )
       else:
            lowerCAmelCase_       :							Tuple			 =		AutoConfig.for_model(_lowercase     ,    **_lowercase    )

       lowerCAmelCase_       :							List[Any]			 =		AutoConfig.for_model(_lowercase     ,    **_lowercase    )

       lowerCAmelCase_       :							Union[str, Any]			 =		projection_dim
       lowerCAmelCase_       :							Optional[int]			 =		logit_scale_init_value







  @classmethod
  def 						UpperCAmelCase__		(						cls     ,    _lowercase     ,    _lowercase     ,    **_lowercase    ):
       return cls(vision_config=vision_config.to_dict()     ,    text_config=text_config.to_dict()     ,    **_lowercase    )







  def 						UpperCAmelCase__		(						self    ):
       lowerCAmelCase_       :							int			 =		copy.deepcopy(self.__dict__    )
       lowerCAmelCase_       :							List[str]			 =		self.vision_config.to_dict()
       lowerCAmelCase_       :							List[str]			 =		self.text_config.to_dict()
       lowerCAmelCase_       :							str			 =		self.__class__.model_type
       return output
