



'''simple docstring'''






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


lowerCAmelCase:					List[str]						       =						logging.get_logger(__name__)  # pylint: disable=invalid-name

lowerCAmelCase:					Optional[int]						       =						"\n    Examples:\n        ```py\n        >>> from diffusers import KandinskyV22Pipeline, KandinskyV22PriorPipeline\n        >>> import torch\n\n        >>> pipe_prior = KandinskyV22PriorPipeline.from_pretrained(\"kandinsky-community/kandinsky-2-2-prior\")\n        >>> pipe_prior.to(\"cuda\")\n        >>> prompt = \"red cat, 4k photo\"\n        >>> out = pipe_prior(prompt)\n        >>> image_emb = out.image_embeds\n        >>> zero_image_emb = out.negative_image_embeds\n        >>> pipe = KandinskyV22Pipeline.from_pretrained(\"kandinsky-community/kandinsky-2-2-decoder\")\n        >>> pipe.to(\"cuda\")\n        >>> image = pipe(\n        ...     image_embeds=image_emb,\n        ...     negative_image_embeds=zero_image_emb,\n        ...     height=768,\n        ...     width=768,\n        ...     num_inference_steps=50,\n        ... ).images\n        >>> image[0].save(\"cat.png\")\n        ```\n"
def       lowerCamelCase__       (						_A ,			_A ,			_A=8				):
  a						:					Optional[Any]        =     height // scale_factor**2
  if height % scale_factor**2 != 0:
    new_height += 1
  a						:					Dict        =     width // scale_factor**2
  if width % scale_factor**2 != 0:
    new_width += 1
  return new_height * scale_factor, new_width * scale_factor

class 							a__(					lowercase__	):







    def __init__(      self :  List[Any]						,			__snake_case :  UNetaDConditionModel						,			__snake_case :  DDPMScheduler						,			__snake_case :  VQModel						,			):
      super().__init__()

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







    def 			lowercase_       (      self :  int						,			__snake_case :  Optional[int]						,			__snake_case :  int						,			__snake_case :  Optional[int]						,			__snake_case :  str						,			__snake_case :  List[str]						,			__snake_case :  str	):
      if latents is None:
        a						:					Any        =     randn_tensor(__lowercase						,			generator=__lowercase						,			device=__lowercase						,			dtype=__lowercase	)
      else:
        if latents.shape != shape:
          raise ValueError(F"""Unexpected latents shape, got {latents.shape}, expected {shape}"""	)
        a						:					int        =     latents.to(__lowercase	)

      a						:					int        =     latents * scheduler.init_noise_sigma
      return latents







    def 			lowercase_       (      self :  Any						,			__snake_case :  int=0	):
      if is_accelerate_available():
        from accelerate import cpu_offload
      else:
        raise ImportError('Please install accelerate via `pip install accelerate`'	)

      a						:					Union[str, Any]        =     torch.device(F"""cuda:{gpu_id}"""	)

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







    def 			lowercase_       (      self :  Union[str, Any]						,			__snake_case :  Dict=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.'	)

      a						:					Dict        =     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						:					str        =     None
      for cpu_offloaded_model in [self.unet, self.movq]:
        a						:					Tuple        =     cpu_offload_with_hook(__lowercase						,			__lowercase						,			prev_module_hook=__lowercase	)

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







    @property
    # Copied from diffusers.pipelines.stable_diffusion.pipeline_stable_diffusion.StableDiffusionPipeline._execution_device
    def 			lowercase_       (      self :  Optional[int]	):
      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







    @torch.no_grad()
    @replace_example_docstring(__lowercase	)
    def __call__(      self :  int						,			__snake_case :  Union[torch.FloatTensor, List[torch.FloatTensor]]						,			__snake_case :  Union[torch.FloatTensor, List[torch.FloatTensor]]						,			__snake_case :  int = 5_12						,			__snake_case :  int = 5_12						,			__snake_case :  int = 1_00						,			__snake_case :  float = 4.0						,			__snake_case :  int = 1						,			__snake_case :  Optional[Union[torch.Generator, List[torch.Generator]]] = None						,			__snake_case :  Optional[torch.FloatTensor] = None						,			__snake_case :  Optional[str] = "pil"						,			__snake_case :  bool = True						,			):
      a						:					Union[str, Any]        =     self._execution_device

      a						:					List[str]        =     guidance_scale > 1.0

      if isinstance(__lowercase						,			__lowercase	):
        a						:					Any        =     torch.cat(__lowercase						,			dim=0	)
      a						:					List[Any]        =     image_embeds.shape[0] * num_images_per_prompt
      if isinstance(__lowercase						,			__lowercase	):
        a						:					List[str]        =     torch.cat(__lowercase						,			dim=0	)

      if do_classifier_free_guidance:
        a						:					Any        =     image_embeds.repeat_interleave(__lowercase						,			dim=0	)
        a						:					Tuple        =     negative_image_embeds.repeat_interleave(__lowercase						,			dim=0	)

      a						:					List[Any]        =     torch.cat([negative_image_embeds, image_embeds]						,			dim=0	).to(dtype=self.unet.dtype						,			device=__lowercase	)

      self.scheduler.set_timesteps(__lowercase						,			device=__lowercase	)
      a						:					str        =     self.scheduler.timesteps

      a						:					Optional[Any]        =     self.unet.config.in_channels

      a						:					Optional[int]        =     downscale_height_and_width(__lowercase						,			__lowercase						,			self.movq_scale_factor	)

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

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

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

        if do_classifier_free_guidance:
          a						:					Optional[int]        =     noise_pred.split(latents.shape[1]						,			dim=1	)
          a						:					Tuple        =     noise_pred.chunk(2	)
          a						:					Tuple        =     variance_pred.chunk(2	)
          a						:					List[str]        =     noise_pred_uncond + guidance_scale * (noise_pred_text - noise_pred_uncond)
          a						:					Optional[Any]        =     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						:					str        =     noise_pred.split(latents.shape[1]						,			dim=1	)

        # compute the previous noisy sample x_t -> x_t-1
        a						:					Tuple        =     self.scheduler.step(
            __lowercase						,			__lowercase						,			__lowercase						,			generator=__lowercase						,			)[0]
      # post-processing
      a						:					Optional[int]        =     self.movq.decode(__lowercase						,			force_not_quantize=__lowercase	)["""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						:					List[str]        =     image * 0.5 + 0.5
        a						:					str        =     image.clamp(0						,			1	)
        a						:					int        =     image.cpu().permute(0						,			2						,			3						,			1	).float().numpy()

      if output_type == "pil":
        a						:					Optional[int]        =     self.numpy_to_pil(__lowercase	)

      if not return_dict:
        return (image,)

      return ImagePipelineOutput(images=__lowercase	)







import random
import unittest

import numpy as np

from diffusers import (
    DPMSolverMultistepScheduler,
    EulerAncestralDiscreteScheduler,
    EulerDiscreteScheduler,
    LMSDiscreteScheduler,
    OnnxStableDiffusionImgaImgPipeline,
    PNDMScheduler,
)
from diffusers.utils import floats_tensor
from diffusers.utils.testing_utils import (
    is_onnx_available,
    load_image,
    nightly,
    require_onnxruntime,
    require_torch_gpu,
)

from ..test_pipelines_onnx_common import OnnxPipelineTesterMixin


if is_onnx_available():
       import onnxruntime as ort
class     a    ( lowercase__							,       unittest.TestCase   ):



      """simple docstring"""


      a  : int							       =	'hf-internal-testing/tiny-random-OnnxStableDiffusionPipeline'
      def 	UpperCAmelCase      (      self					:			List[str]		,      __lowercase					:			Optional[Any]=0					)							->      Any:
        __UpperCAmelCase							:	Any           =  floats_tensor((1, 3, 128, 128)		,      rng=random.Random(__lowercase					)					)
        __UpperCAmelCase							:	int           =  np.random.RandomState(__lowercase					)
        __UpperCAmelCase							:	Optional[Any]           =  {
            """prompt""": """A painting of a squirrel eating a burger""",
            """image""": image,
            """generator""": generator,
            """num_inference_steps""": 3,
            """strength""": 0.75,
            """guidance_scale""": 7.5,
            """output_type""": """numpy""",
        }
        return inputs
      def 	UpperCAmelCase      (      self					:			Optional[int]					)							->      Optional[Any]:
        __UpperCAmelCase							:	List[Any]           =  OnnxStableDiffusionImgaImgPipeline.from_pretrained(self.hub_checkpoint		,      provider="""CPUExecutionProvider"""					)
        pipe.set_progress_bar_config(disable=__lowercase					)

        __UpperCAmelCase							:	int           =  self.get_dummy_inputs()
        __UpperCAmelCase							:	Optional[Any]           =  pipe(**__lowercase					).images
        __UpperCAmelCase							:	Union[str, Any]           =  image[0, -3:, -3:, -1].flatten()

        assert image.shape == (1, 128, 128, 3)
        __UpperCAmelCase							:	List[str]           =  np.array([0.69_643, 0.58_484, 0.50_314, 0.58_760, 0.55_368, 0.59_643, 0.51_529, 0.41_217, 0.49_087]					)
        assert np.abs(image_slice - expected_slice					).max() < 1e-1
      def 	UpperCAmelCase      (      self					:			Optional[Any]					)							->      List[str]:
        __UpperCAmelCase							:	Dict           =  OnnxStableDiffusionImgaImgPipeline.from_pretrained(self.hub_checkpoint		,      provider="""CPUExecutionProvider"""					)
        __UpperCAmelCase							:	int           =  PNDMScheduler.from_config(pipe.scheduler.config		,      skip_prk_steps=__lowercase					)
        pipe.set_progress_bar_config(disable=__lowercase					)

        __UpperCAmelCase							:	Any           =  self.get_dummy_inputs()
        __UpperCAmelCase							:	Tuple           =  pipe(**__lowercase					).images
        __UpperCAmelCase							:	Dict           =  image[0, -3:, -3:, -1]

        assert image.shape == (1, 128, 128, 3)
        __UpperCAmelCase							:	str           =  np.array([0.61_737, 0.54_642, 0.53_183, 0.54_465, 0.52_742, 0.60_525, 0.49_969, 0.40_655, 0.48_154]					)

        assert np.abs(image_slice.flatten() - expected_slice					).max() < 1e-1
      def 	UpperCAmelCase      (      self					:			str					)							->      Tuple:
        __UpperCAmelCase							:	Tuple           =  OnnxStableDiffusionImgaImgPipeline.from_pretrained(self.hub_checkpoint		,      provider="""CPUExecutionProvider"""					)
        __UpperCAmelCase							:	int           =  LMSDiscreteScheduler.from_config(pipe.scheduler.config					)
        pipe.set_progress_bar_config(disable=__lowercase					)

        # warmup pass to apply optimizations
        __UpperCAmelCase							:	Optional[int]           =  pipe(**self.get_dummy_inputs()					)

        __UpperCAmelCase							:	Tuple           =  self.get_dummy_inputs()
        __UpperCAmelCase							:	Any           =  pipe(**__lowercase					).images
        __UpperCAmelCase							:	List[Any]           =  image[0, -3:, -3:, -1]

        assert image.shape == (1, 128, 128, 3)
        __UpperCAmelCase							:	Optional[int]           =  np.array([0.52_761, 0.59_977, 0.49_033, 0.49_619, 0.54_282, 0.50_311, 0.47_600, 0.40_918, 0.45_203]					)

        assert np.abs(image_slice.flatten() - expected_slice					).max() < 1e-1
      def 	UpperCAmelCase      (      self					:			Optional[Any]					)							->      str:
        __UpperCAmelCase							:	int           =  OnnxStableDiffusionImgaImgPipeline.from_pretrained(self.hub_checkpoint		,      provider="""CPUExecutionProvider"""					)
        __UpperCAmelCase							:	Union[str, Any]           =  EulerDiscreteScheduler.from_config(pipe.scheduler.config					)
        pipe.set_progress_bar_config(disable=__lowercase					)

        __UpperCAmelCase							:	List[str]           =  self.get_dummy_inputs()
        __UpperCAmelCase							:	int           =  pipe(**__lowercase					).images
        __UpperCAmelCase							:	Dict           =  image[0, -3:, -3:, -1]

        assert image.shape == (1, 128, 128, 3)
        __UpperCAmelCase							:	Tuple           =  np.array([0.52_911, 0.60_004, 0.49_229, 0.49_805, 0.54_502, 0.50_680, 0.47_777, 0.41_028, 0.45_304]					)

        assert np.abs(image_slice.flatten() - expected_slice					).max() < 1e-1
      def 	UpperCAmelCase      (      self					:			int					)							->      Any:
        __UpperCAmelCase							:	Any           =  OnnxStableDiffusionImgaImgPipeline.from_pretrained(self.hub_checkpoint		,      provider="""CPUExecutionProvider"""					)
        __UpperCAmelCase							:	List[Any]           =  EulerAncestralDiscreteScheduler.from_config(pipe.scheduler.config					)
        pipe.set_progress_bar_config(disable=__lowercase					)

        __UpperCAmelCase							:	Union[str, Any]           =  self.get_dummy_inputs()
        __UpperCAmelCase							:	int           =  pipe(**__lowercase					).images
        __UpperCAmelCase							:	List[Any]           =  image[0, -3:, -3:, -1]

        assert image.shape == (1, 128, 128, 3)
        __UpperCAmelCase							:	List[str]           =  np.array([0.52_911, 0.60_004, 0.49_229, 0.49_805, 0.54_502, 0.50_680, 0.47_777, 0.41_028, 0.45_304]					)

        assert np.abs(image_slice.flatten() - expected_slice					).max() < 1e-1




      def 	UpperCAmelCase      (      self					:			Tuple					)							->      str:
        __UpperCAmelCase							:	Dict           =  OnnxStableDiffusionImgaImgPipeline.from_pretrained(self.hub_checkpoint		,      provider="""CPUExecutionProvider"""					)
        __UpperCAmelCase							:	str           =  DPMSolverMultistepScheduler.from_config(pipe.scheduler.config					)
        pipe.set_progress_bar_config(disable=__lowercase					)

        __UpperCAmelCase							:	Optional[Any]           =  self.get_dummy_inputs()
        __UpperCAmelCase							:	int           =  pipe(**__lowercase					).images
        __UpperCAmelCase							:	List[Any]           =  image[0, -3:, -3:, -1]

        assert image.shape == (1, 128, 128, 3)
        __UpperCAmelCase							:	Union[str, Any]           =  np.array([0.65_331, 0.58_277, 0.48_204, 0.56_059, 0.53_665, 0.56_235, 0.50_969, 0.40_009, 0.46_552]					)

        assert np.abs(image_slice.flatten() - expected_slice					).max() < 1e-1


@nightly
@require_onnxruntime
@require_torch_gpu
class     a    ( unittest.TestCase   ):



      """simple docstring"""


      @property
      def 	UpperCAmelCase      (      self					:			Dict					)							->      List[Any]:
        return (
            "CUDAExecutionProvider",
            {
                "gpu_mem_limit": "15000000000",  # 15GB
                "arena_extend_strategy": "kSameAsRequested",
            },
        )
      @property
      def 	UpperCAmelCase      (      self					:			Tuple					)							->      Tuple:
        __UpperCAmelCase							:	Optional[int]           =  ort.SessionOptions()
        __UpperCAmelCase							:	List[Any]           =  False
        return options
      def 	UpperCAmelCase      (      self					:			List[str]					)							->      Tuple:
        __UpperCAmelCase							:	List[Any]           =  load_image(
            """https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main"""
            """/img2img/sketch-mountains-input.jpg"""					)
        __UpperCAmelCase							:	Dict           =  init_image.resize((768, 512)					)
        # using the PNDM scheduler by default
        __UpperCAmelCase							:	Any           =  OnnxStableDiffusionImgaImgPipeline.from_pretrained(
            """CompVis/stable-diffusion-v1-4"""		,      revision="""onnx"""		,      safety_checker=__lowercase		,      feature_extractor=__lowercase		,      provider=self.gpu_provider		,      sess_options=self.gpu_options		,      )
        pipe.set_progress_bar_config(disable=__lowercase					)

        __UpperCAmelCase							:	Dict           =  """A fantasy landscape, trending on artstation"""

        __UpperCAmelCase							:	str           =  np.random.RandomState(0					)
        __UpperCAmelCase							:	Optional[Any]           =  pipe(
            prompt=__lowercase		,      image=__lowercase		,      strength=0.75		,      guidance_scale=7.5		,      num_inference_steps=10		,      generator=__lowercase		,      output_type="""np"""		,      )
        __UpperCAmelCase							:	str           =  output.images
        __UpperCAmelCase							:	int           =  images[0, 255:258, 383:386, -1]

        assert images.shape == (1, 512, 768, 3)
        __UpperCAmelCase							:	Union[str, Any]           =  np.array([0.4_909, 0.5_059, 0.5_372, 0.4_623, 0.4_876, 0.5_049, 0.4_820, 0.4_956, 0.5_019]					)
        # TODO: lower the tolerance after finding the cause of onnxruntime reproducibility issues

        assert np.abs(image_slice.flatten() - expected_slice					).max() < 2e-2




      def 	UpperCAmelCase      (      self					:			Optional[Any]					)							->      str:
        __UpperCAmelCase							:	Tuple           =  load_image(
            """https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main"""
            """/img2img/sketch-mountains-input.jpg"""					)
        __UpperCAmelCase							:	int           =  init_image.resize((768, 512)					)
        __UpperCAmelCase							:	Tuple           =  LMSDiscreteScheduler.from_pretrained(
            """runwayml/stable-diffusion-v1-5"""		,      subfolder="""scheduler"""		,      revision="""onnx"""					)
        __UpperCAmelCase							:	Tuple           =  OnnxStableDiffusionImgaImgPipeline.from_pretrained(
            """runwayml/stable-diffusion-v1-5"""		,      revision="""onnx"""		,      scheduler=__lowercase		,      safety_checker=__lowercase		,      feature_extractor=__lowercase		,      provider=self.gpu_provider		,      sess_options=self.gpu_options		,      )
        pipe.set_progress_bar_config(disable=__lowercase					)

        __UpperCAmelCase							:	Dict           =  """A fantasy landscape, trending on artstation"""

        __UpperCAmelCase							:	int           =  np.random.RandomState(0					)
        __UpperCAmelCase							:	Optional[int]           =  pipe(
            prompt=__lowercase		,      image=__lowercase		,      strength=0.75		,      guidance_scale=7.5		,      num_inference_steps=20		,      generator=__lowercase		,      output_type="""np"""		,      )
        __UpperCAmelCase							:	Union[str, Any]           =  output.images
        __UpperCAmelCase							:	Union[str, Any]           =  images[0, 255:258, 383:386, -1]

        assert images.shape == (1, 512, 768, 3)
        __UpperCAmelCase							:	str           =  np.array([0.8_043, 0.926, 0.9_581, 0.8_119, 0.8_954, 0.913, 0.7_209, 0.7_463, 0.7_431]					)
        # TODO: lower the tolerance after finding the cause of onnxruntime reproducibility issues

        assert np.abs(image_slice.flatten() - expected_slice					).max() < 2e-2


