


from typing import List, Optional, Union

import torch

from ...models import UNetaDConditionModel, VQModel
from ...pipelines import DiffusionPipeline
from ...pipelines.pipeline_utils import ImagePipelineOutput
from ...schedulers import DDPMScheduler
from ...utils import (
    is_accelerate_available,
    is_accelerate_version,
    logging,
    randn_tensor,
    replace_example_docstring,
)


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

A_		:     Dict        =     '''
    Examples:
        ```py
        >>> from diffusers import KandinskyV22Pipeline, KandinskyV22PriorPipeline
        >>> import torch

        >>> pipe_prior = KandinskyV22PriorPipeline.from_pretrained("kandinsky-community/kandinsky-2-2-prior")
        >>> pipe_prior.to("cuda")
        >>> prompt = "red cat, 4k photo"
        >>> out = pipe_prior(prompt)
        >>> image_emb = out.image_embeds
        >>> zero_image_emb = out.negative_image_embeds
        >>> pipe = KandinskyV22Pipeline.from_pretrained("kandinsky-community/kandinsky-2-2-decoder")
        >>> pipe.to("cuda")
        >>> image = pipe(
        ...     image_embeds=image_emb,
        ...     negative_image_embeds=zero_image_emb,
        ...     height=768,
        ...     width=768,
        ...     num_inference_steps=50,
        ... ).images
        >>> image[0].save("cat.png")
        ```
'''


def 	UpperCAmelCase__      (      UpperCAmelCase__ :Optional[Any] ,					UpperCAmelCase__ :Optional[int] ,					UpperCAmelCase__ :Optional[int]=8					):






 '''simple docstring'''

 a						=		height // scale_factor**2
 if height % scale_factor**2 != 0:
  new_height += 1
 a						=		width // scale_factor**2
 if width % scale_factor**2 != 0:
  new_width += 1
 return new_height * scale_factor, new_width * scale_factor




class 		_lowercase		(       UpperCAmelCase__	):

   def __init__(      self			:       Optional[Any]							,   __lowerCAmelCase			:       UNetaDConditionModel							,   __lowerCAmelCase			:       DDPMScheduler							,   __lowerCAmelCase			:       VQModel							,   )       ->				Union[str, Any]:


    """simple docstring"""






    super().__init__()

    self.register_modules(
        unet=__lowerCAmelCase							,   scheduler=__lowerCAmelCase							,   movq=__lowerCAmelCase							,   )
    a						=		2 ** (len(self.movq.config.block_out_channels							) - 1)

   def 		A							(      self			:       str							,   __lowerCAmelCase			:       Optional[Any]							,   __lowerCAmelCase			:       Optional[Any]							,   __lowerCAmelCase			:       List[str]							,   __lowerCAmelCase			:       Optional[Any]							,   __lowerCAmelCase			:       Tuple							,   __lowerCAmelCase			:       Dict							)       ->				Optional[int]:


    """simple docstring"""






    if latents is None:
     a						=		randn_tensor(__lowerCAmelCase							,   generator=__lowerCAmelCase							,   device=__lowerCAmelCase							,   dtype=__lowerCAmelCase							)
    else:
     if latents.shape != shape:
      raise ValueError(f"""Unexpected latents shape, got {latents.shape}, expected {shape}"""							)
     a						=		latents.to(__lowerCAmelCase							)

    a						=		latents * scheduler.init_noise_sigma
    return latents

   def 		A							(      self			:       Any							,   __lowerCAmelCase			:       Any=0							)       ->				List[Any]:


    """simple docstring"""






    if is_accelerate_available():
     from accelerate import cpu_offload
    else:
     raise ImportError("Please install accelerate via `pip install accelerate`"							)

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

    a						=		[
        self.unet,
        self.movq,
    ]
    for cpu_offloaded_model in models:
     if cpu_offloaded_model is not None:
      cpu_offload(__lowerCAmelCase							,   __lowerCAmelCase							)

   def 		A							(      self			:       Dict							,   __lowerCAmelCase			:       Any=0							)       ->				Any:


    """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=__lowerCAmelCase							)
     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, self.movq]:
     a		,  a						=		cpu_offload_with_hook(__lowerCAmelCase							,   __lowerCAmelCase							,   prev_module_hook=__lowerCAmelCase							)

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

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


    """simple docstring"""






    if not hasattr(self.unet							,   "_hf_hook"							):
     return self.device
    for module in self.unet.modules():
     if (
         hasattr(__lowerCAmelCase							,   "_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




   @torch.no_grad()
   @replace_example_docstring(__lowerCAmelCase							)
   def __call__(      self			:       List[str]							,   __lowerCAmelCase			:       Union[torch.FloatTensor, List[torch.FloatTensor]]							,   __lowerCAmelCase			:       Union[torch.FloatTensor, List[torch.FloatTensor]]							,   __lowerCAmelCase			:       int = 512							,   __lowerCAmelCase			:       int = 512							,   __lowerCAmelCase			:       int = 100							,   __lowerCAmelCase			:       float = 4.0							,   __lowerCAmelCase			:       int = 1							,   __lowerCAmelCase			:       Optional[Union[torch.Generator, List[torch.Generator]]] = None							,   __lowerCAmelCase			:       Optional[torch.FloatTensor] = None							,   __lowerCAmelCase			:       Optional[str] = "pil"							,   __lowerCAmelCase			:       bool = True							,   )       ->				List[str]:


    """simple docstring"""






    a						=		self._execution_device

    a						=		guidance_scale > 1.0

    if isinstance(__lowerCAmelCase							,   __lowerCAmelCase							):
     a						=		torch.cat(__lowerCAmelCase							,   dim=0							)
    a						=		image_embeds.shape[0] * num_images_per_prompt
    if isinstance(__lowerCAmelCase							,   __lowerCAmelCase							):
     a						=		torch.cat(__lowerCAmelCase							,   dim=0							)

    if do_classifier_free_guidance:
     a						=		image_embeds.repeat_interleave(__lowerCAmelCase							,   dim=0							)
     a						=		negative_image_embeds.repeat_interleave(__lowerCAmelCase							,   dim=0							)

    a						=		torch.cat([negative_image_embeds, image_embeds]							,   dim=0							).to(dtype=self.unet.dtype							,   device=__lowerCAmelCase							)

    self.scheduler.set_timesteps(__lowerCAmelCase							,   device=__lowerCAmelCase							)
    a						=		self.scheduler.timesteps

    a						=		self.unet.config.in_channels

    a		,  a						=		downscale_height_and_width(__lowerCAmelCase							,   __lowerCAmelCase							,   self.movq_scale_factor							)

    # create initial latent
    a						=		self.prepare_latents(
        (batch_size, num_channels_latents, height, width)							,   image_embeds.dtype							,   __lowerCAmelCase							,   __lowerCAmelCase							,   __lowerCAmelCase							,   self.scheduler							,   )

    for i, t in enumerate(self.progress_bar(__lowerCAmelCase							)							):
     # expand the latents if we are doing classifier free guidance
     a						=		torch.cat([latents] * 2							) if do_classifier_free_guidance else latents

     a						=		{"image_embeds": image_embeds}
     a						=		self.unet(
         sample=__lowerCAmelCase							,   timestep=__lowerCAmelCase							,   encoder_hidden_states=__lowerCAmelCase							,   added_cond_kwargs=__lowerCAmelCase							,   return_dict=__lowerCAmelCase							,   )[0]

     if do_classifier_free_guidance:
      a		,  a						=		noise_pred.split(latents.shape[1]							,   dim=1							)
      a		,  a						=		noise_pred.chunk(2							)
      a		,  a						=		variance_pred.chunk(2							)
      a						=		noise_pred_uncond + guidance_scale * (noise_pred_text - noise_pred_uncond)
      a						=		torch.cat([noise_pred, variance_pred_text]							,   dim=1							)

     if not (
         hasattr(self.scheduler.config							,   "variance_type"							)
         and self.scheduler.config.variance_type in ["learned", "learned_range"]
     ):
      a		,  a						=		noise_pred.split(latents.shape[1]							,   dim=1							)

     # compute the previous noisy sample x_t -> x_t-1
     a						=		self.scheduler.step(
         __lowerCAmelCase							,   __lowerCAmelCase							,   __lowerCAmelCase							,   generator=__lowerCAmelCase							,   )[0]
    # post-processing
    a						=		self.movq.decode(__lowerCAmelCase							,   force_not_quantize=__lowerCAmelCase							)["sample"]

    if output_type not in ["pt", "np", "pil"]:
     raise ValueError(f"""Only the output types `pt`, `pil` and `np` are supported not output_type={output_type}"""							)

    if output_type in ["np", "pil"]:
     a						=		image * 0.5 + 0.5
     a						=		image.clamp(0							,   1							)
     a						=		image.cpu().permute(0							,   2							,   3							,   1							).float().numpy()

    if output_type == "pil":
     a						=		self.numpy_to_pil(__lowerCAmelCase							)

    if not return_dict:
     return (image,)

    return ImagePipelineOutput(images=__lowerCAmelCase							)






import tempfile

import torch

from diffusers import (
    DEISMultistepScheduler,
    DPMSolverMultistepScheduler,
    DPMSolverSinglestepScheduler,
    UniPCMultistepScheduler,
)

from .test_schedulers import SchedulerCommonTest



class 		_lowercase		(       UpperCAmelCase__	):
   _UpperCAmelCase     	=			(UniPCMultistepScheduler,)
   _UpperCAmelCase     	=			(('''num_inference_steps''', 25),)

   def 		A							(      self			:       List[Any]							,   **__lowerCAmelCase			:       Optional[int]							)       ->				int:


    """simple docstring"""






    a						=		{
        "num_train_timesteps": 1000,
        "beta_start": 0.0_0_0_1,
        "beta_end": 0.0_2,
        "beta_schedule": "linear",
        "solver_order": 2,
        "solver_type": "bh2",
    }

    config.update(**__lowerCAmelCase							)
    return config

   def 		A							(      self			:       List[Any]							,   __lowerCAmelCase			:       Optional[int]=0							,   **__lowerCAmelCase			:       Optional[Any]							)       ->				int:


    """simple docstring"""






    a						=		dict(self.forward_default_kwargs							)
    a						=		kwargs.pop("num_inference_steps"							,   __lowerCAmelCase							)
    a						=		self.dummy_sample
    a						=		0.1 * sample
    a						=		[residual + 0.2, residual + 0.1_5, residual + 0.1_0]

    for scheduler_class in self.scheduler_classes:
     a						=		self.get_scheduler_config(**__lowerCAmelCase							)
     a						=		scheduler_class(**__lowerCAmelCase							)
     scheduler.set_timesteps(__lowerCAmelCase							)
     # copy over dummy past residuals
     a						=		dummy_past_residuals[: scheduler.config.solver_order]

     with tempfile.TemporaryDirectory() as tmpdirname:
      scheduler.save_config(__lowerCAmelCase							)
      a						=		scheduler_class.from_pretrained(__lowerCAmelCase							)
      new_scheduler.set_timesteps(__lowerCAmelCase							)
      # copy over dummy past residuals
      a						=		dummy_past_residuals[: new_scheduler.config.solver_order]

     a		,  a						=		sample, sample
     for t in range(__lowerCAmelCase							,   time_step + scheduler.config.solver_order + 1							):
      a						=		scheduler.step(__lowerCAmelCase							,   __lowerCAmelCase							,   __lowerCAmelCase							,   **__lowerCAmelCase							).prev_sample
      a						=		new_scheduler.step(__lowerCAmelCase							,   __lowerCAmelCase							,   __lowerCAmelCase							,   **__lowerCAmelCase							).prev_sample

      assert torch.sum(torch.abs(output - new_output							)							) < 1E-5, "Scheduler outputs are not identical"

   def 		A							(      self			:       List[Any]							,   __lowerCAmelCase			:       Optional[Any]=0							,   **__lowerCAmelCase			:       List[Any]							)       ->				List[str]:


    """simple docstring"""






    a						=		dict(self.forward_default_kwargs							)
    a						=		kwargs.pop("num_inference_steps"							,   __lowerCAmelCase							)
    a						=		self.dummy_sample
    a						=		0.1 * sample
    a						=		[residual + 0.2, residual + 0.1_5, residual + 0.1_0]

    for scheduler_class in self.scheduler_classes:
     a						=		self.get_scheduler_config()
     a						=		scheduler_class(**__lowerCAmelCase							)
     scheduler.set_timesteps(__lowerCAmelCase							)

     # copy over dummy past residuals (must be after setting timesteps)
     a						=		dummy_past_residuals[: scheduler.config.solver_order]

     with tempfile.TemporaryDirectory() as tmpdirname:
      scheduler.save_config(__lowerCAmelCase							)
      a						=		scheduler_class.from_pretrained(__lowerCAmelCase							)
      # copy over dummy past residuals
      new_scheduler.set_timesteps(__lowerCAmelCase							)

      # copy over dummy past residual (must be after setting timesteps)
      a						=		dummy_past_residuals[: new_scheduler.config.solver_order]

     a						=		scheduler.step(__lowerCAmelCase							,   __lowerCAmelCase							,   __lowerCAmelCase							,   **__lowerCAmelCase							).prev_sample
     a						=		new_scheduler.step(__lowerCAmelCase							,   __lowerCAmelCase							,   __lowerCAmelCase							,   **__lowerCAmelCase							).prev_sample

     assert torch.sum(torch.abs(output - new_output							)							) < 1E-5, "Scheduler outputs are not identical"

   def 		A							(      self			:       str							,   __lowerCAmelCase			:       Any=None							,   **__lowerCAmelCase			:       List[str]							)       ->				Any:


    """simple docstring"""






    if scheduler is None:
     a						=		self.scheduler_classes[0]
     a						=		self.get_scheduler_config(**__lowerCAmelCase							)
     a						=		scheduler_class(**__lowerCAmelCase							)

    a						=		self.scheduler_classes[0]
    a						=		self.get_scheduler_config(**__lowerCAmelCase							)
    a						=		scheduler_class(**__lowerCAmelCase							)

    a						=		10
    a						=		self.dummy_model()
    a						=		self.dummy_sample_deter
    scheduler.set_timesteps(__lowerCAmelCase							)

    for i, t in enumerate(scheduler.timesteps							):
     a						=		model(__lowerCAmelCase							,   __lowerCAmelCase							)
     a						=		scheduler.step(__lowerCAmelCase							,   __lowerCAmelCase							,   __lowerCAmelCase							).prev_sample

    return sample

   def 		A							(      self			:       Any							)       ->				int:


    """simple docstring"""






    a						=		dict(self.forward_default_kwargs							)

    a						=		kwargs.pop("num_inference_steps"							,   __lowerCAmelCase							)

    for scheduler_class in self.scheduler_classes:
     a						=		self.get_scheduler_config()
     a						=		scheduler_class(**__lowerCAmelCase							)

     a						=		self.dummy_sample
     a						=		0.1 * sample

     if num_inference_steps is not None and hasattr(__lowerCAmelCase							,   "set_timesteps"							):
      scheduler.set_timesteps(__lowerCAmelCase							)
     elif num_inference_steps is not None and not hasattr(__lowerCAmelCase							,   "set_timesteps"							):
      a						=		num_inference_steps

     # copy over dummy past residuals (must be done after set_timesteps)
     a						=		[residual + 0.2, residual + 0.1_5, residual + 0.1_0]
     a						=		dummy_past_residuals[: scheduler.config.solver_order]

     a						=		scheduler.timesteps[5]
     a						=		scheduler.timesteps[6]

     a						=		scheduler.step(__lowerCAmelCase							,   __lowerCAmelCase							,   __lowerCAmelCase							,   **__lowerCAmelCase							).prev_sample
     a						=		scheduler.step(__lowerCAmelCase							,   __lowerCAmelCase							,   __lowerCAmelCase							,   **__lowerCAmelCase							).prev_sample

     self.assertEqual(output_a.shape							,   sample.shape							)
     self.assertEqual(output_a.shape							,   output_a.shape							)

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


    """simple docstring"""






    a						=		UniPCMultistepScheduler(**self.get_scheduler_config()							)
    a						=		self.full_loop(scheduler=__lowerCAmelCase							)
    a						=		torch.mean(torch.abs(__lowerCAmelCase							)							)

    assert abs(result_mean.item() - 0.2_4_6_4							) < 1E-3

    a						=		DPMSolverSinglestepScheduler.from_config(scheduler.config							)
    a						=		DEISMultistepScheduler.from_config(scheduler.config							)
    a						=		DPMSolverMultistepScheduler.from_config(scheduler.config							)
    a						=		UniPCMultistepScheduler.from_config(scheduler.config							)

    a						=		self.full_loop(scheduler=__lowerCAmelCase							)
    a						=		torch.mean(torch.abs(__lowerCAmelCase							)							)

    assert abs(result_mean.item() - 0.2_4_6_4							) < 1E-3

   def 		A							(      self			:       List[Any]							)       ->				Dict:


    """simple docstring"""






    for timesteps in [25, 50, 100, 999, 1000]:
     self.check_over_configs(num_train_timesteps=__lowerCAmelCase							)

   def 		A							(      self			:       Optional[Any]							)       ->				Tuple:


    """simple docstring"""






    self.check_over_configs(thresholding=__lowerCAmelCase							)
    for order in [1, 2, 3]:
     for solver_type in ["bh1", "bh2"]:
      for threshold in [0.5, 1.0, 2.0]:
       for prediction_type in ["epsilon", "sample"]:
        self.check_over_configs(
            thresholding=__lowerCAmelCase							,   prediction_type=__lowerCAmelCase							,   sample_max_value=__lowerCAmelCase							,   solver_order=__lowerCAmelCase							,   solver_type=__lowerCAmelCase							,   )

   def 		A							(      self			:       Optional[Any]							)       ->				Any:


    """simple docstring"""






    for prediction_type in ["epsilon", "v_prediction"]:
     self.check_over_configs(prediction_type=__lowerCAmelCase							)

   def 		A							(      self			:       Optional[Any]							)       ->				Any:


    """simple docstring"""






    for solver_type in ["bh1", "bh2"]:
     for order in [1, 2, 3]:
      for prediction_type in ["epsilon", "sample"]:
       self.check_over_configs(
           solver_order=__lowerCAmelCase							,   solver_type=__lowerCAmelCase							,   prediction_type=__lowerCAmelCase							,   )
       a						=		self.full_loop(
           solver_order=__lowerCAmelCase							,   solver_type=__lowerCAmelCase							,   prediction_type=__lowerCAmelCase							,   )
       assert not torch.isnan(__lowerCAmelCase							).any(), "Samples have nan numbers"

   def 		A							(      self			:       Optional[int]							)       ->				Any:


    """simple docstring"""






    self.check_over_configs(lower_order_final=__lowerCAmelCase							)
    self.check_over_configs(lower_order_final=__lowerCAmelCase							)

   def 		A							(      self			:       Dict							)       ->				str:


    """simple docstring"""






    for num_inference_steps in [1, 2, 3, 5, 10, 50, 100, 999, 1000]:
     self.check_over_forward(num_inference_steps=__lowerCAmelCase							,   time_step=0							)

   def 		A							(      self			:       Dict							)       ->				int:


    """simple docstring"""






    a						=		self.full_loop()
    a						=		torch.mean(torch.abs(__lowerCAmelCase							)							)

    assert abs(result_mean.item() - 0.2_4_6_4							) < 1E-3

   def 		A							(      self			:       Optional[int]							)       ->				int:


    """simple docstring"""






    a						=		self.full_loop(prediction_type="v_prediction"							)
    a						=		torch.mean(torch.abs(__lowerCAmelCase							)							)

    assert abs(result_mean.item() - 0.1_0_1_4							) < 1E-3

   def 		A							(      self			:       Union[str, Any]							)       ->				str:


    """simple docstring"""






    a						=		self.scheduler_classes[0]
    a						=		self.get_scheduler_config(thresholding=__lowerCAmelCase							,   dynamic_thresholding_ratio=0							)
    a						=		scheduler_class(**__lowerCAmelCase							)

    a						=		10
    a						=		self.dummy_model()
    a						=		self.dummy_sample_deter.half()
    scheduler.set_timesteps(__lowerCAmelCase							)

    for i, t in enumerate(scheduler.timesteps							):
     a						=		model(__lowerCAmelCase							,   __lowerCAmelCase							)
     a						=		scheduler.step(__lowerCAmelCase							,   __lowerCAmelCase							,   __lowerCAmelCase							).prev_sample

    assert sample.dtype == torch.floataa




   def 		A							(      self			:       List[str]							,   **__lowerCAmelCase			:       int							)       ->				Dict:


    """simple docstring"""






    for scheduler_class in self.scheduler_classes:
     a						=		self.get_scheduler_config(**__lowerCAmelCase							)
     a						=		scheduler_class(**__lowerCAmelCase							)

     scheduler.set_timesteps(scheduler.config.num_train_timesteps							)
     assert len(scheduler.timesteps.unique()							) == scheduler.num_inference_steps

