



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

_lowerCAmelCase	:   Optional[Any]    =   "\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     UpperCAmelCase_		(					snake_case__    ,			snake_case__    ,			snake_case__=8					)	->     Optional[Any]:



				"""simple docstring"""





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


class 	__snake_case   ( SCREAMING_SNAKE_CASE ):
		def __init__(							self    ,a_    ,a_    ,a_    ,):







						"""simple docstring"""
						super().__init__()

						self.register_modules(
						    unet=a_    ,scheduler=a_    ,movq=a_    ,)
						lowerCAmelCase__		  =  2 ** (len(self.movq.config.block_out_channels						) - 1)
		def 						SCREAMING_SNAKE_CASE_					(							self    ,a_    ,a_    ,a_    ,a_    ,a_    ,a_						):







						"""simple docstring"""
						if latents is None:
										lowerCAmelCase__		  =  randn_tensor(a_    ,generator=a_    ,device=a_    ,dtype=a_						)
						else:
										if latents.shape != shape:
														raise ValueError(f'Unexpected latents shape, got {latents.shape}, expected {shape}'						)
										lowerCAmelCase__		  =  latents.to(a_						)

						lowerCAmelCase__		  =  latents * scheduler.init_noise_sigma
						return latents
		def 						SCREAMING_SNAKE_CASE_					(							self    ,a_=0						):







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

						lowerCAmelCase__		  =  torch.device(f'cuda:{gpu_id}'						)

						lowerCAmelCase__		  =  [
						    self.unet,
						    self.movq,
						]
						for cpu_offloaded_model in models:
										if cpu_offloaded_model is not None:
														cpu_offload(a_    ,a_						)
		def 						SCREAMING_SNAKE_CASE_					(							self    ,a_=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.'						)

						lowerCAmelCase__		  =  torch.device(f'cuda:{gpu_id}'						)

						if self.device.type != "cpu":
										self.to('cpu'    ,silence_dtype_warnings=a_						)
										torch.cuda.empty_cache()  # otherwise we don't see the memory savings (but they probably exist)

						lowerCAmelCase__		  =  None
						for cpu_offloaded_model in [self.unet, self.movq]:
										lowerCAmelCase__  ,   lowerCAmelCase__		  =  cpu_offload_with_hook(a_    ,a_    ,prev_module_hook=a_						)

						# We'll offload the last model manually.
						lowerCAmelCase__		  =  hook
		@property
		# Copied from diffusers.pipelines.stable_diffusion.pipeline_stable_diffusion.StableDiffusionPipeline._execution_device
		def 						SCREAMING_SNAKE_CASE_					(							self						):







						"""simple docstring"""
						if not hasattr(self.unet    ,'_hf_hook'						):
										return self.device
						for module in self.unet.modules():
										if (
										    hasattr(a_    ,'_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(a_						)
		def __call__(							self    ,a_    ,a_    ,a_ = 512    ,a_ = 512    ,a_ = 100    ,a_ = 4.0    ,a_ = 1    ,a_ = None    ,a_ = None    ,a_ = "pil"    ,a_ = True    ,):







						"""simple docstring"""
						lowerCAmelCase__		  =  self._execution_device

						lowerCAmelCase__		  =  guidance_scale > 1.0

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

						if do_classifier_free_guidance:
										lowerCAmelCase__		  =  image_embeds.repeat_interleave(a_    ,dim=0						)
										lowerCAmelCase__		  =  negative_image_embeds.repeat_interleave(a_    ,dim=0						)

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

						self.scheduler.set_timesteps(a_    ,device=a_						)
						lowerCAmelCase__		  =  self.scheduler.timesteps

						lowerCAmelCase__		  =  self.unet.config.in_channels

						lowerCAmelCase__  ,   lowerCAmelCase__		  =  downscale_height_and_width(a_    ,a_    ,self.movq_scale_factor						)

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

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

										lowerCAmelCase__		  =  {'image_embeds': image_embeds}
										lowerCAmelCase__		  =  self.unet(
										    sample=a_    ,timestep=a_    ,encoder_hidden_states=a_    ,added_cond_kwargs=a_    ,return_dict=a_    ,)[0]

										if do_classifier_free_guidance:
														lowerCAmelCase__  ,   lowerCAmelCase__		  =  noise_pred.split(latents.shape[1]    ,dim=1						)
														lowerCAmelCase__  ,   lowerCAmelCase__		  =  noise_pred.chunk(2						)
														lowerCAmelCase__  ,   lowerCAmelCase__		  =  variance_pred.chunk(2						)
														lowerCAmelCase__		  =  noise_pred_uncond + guidance_scale * (noise_pred_text - noise_pred_uncond)
														lowerCAmelCase__		  =  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"]
										):
														lowerCAmelCase__  ,   lowerCAmelCase__		  =  noise_pred.split(latents.shape[1]    ,dim=1						)

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

						if output_type == "pil":
										lowerCAmelCase__		  =  self.numpy_to_pil(a_						)

						if not return_dict:
										return (image,)

						return ImagePipelineOutput(images=a_						)





import gc
import tempfile
import unittest

import numpy as np
import torch

from diffusers import VersatileDiffusionPipeline
from diffusers.utils.testing_utils import load_image, nightly, require_torch_gpu, torch_device


_lowerCAmelCase	:   Any    =   False


class 	__snake_case   ( unittest.TestCase ):
		pass



@nightly
@require_torch_gpu
class 	__snake_case   ( unittest.TestCase ):
		def 						SCREAMING_SNAKE_CASE_					(							self						):







						"""simple docstring"""
						# clean up the VRAM after each test
						super().tearDown()
						gc.collect()
						torch.cuda.empty_cache()
		def 						SCREAMING_SNAKE_CASE_					(							self						):







						"""simple docstring"""
						lowerCAmelCase__		  =  VersatileDiffusionPipeline.from_pretrained('shi-labs/versatile-diffusion'    ,torch_dtype=torch.floataa						)
						pipe.to(a_						)
						pipe.set_progress_bar_config(disable=a_						)

						lowerCAmelCase__		  =  load_image(
						    'https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/versatile_diffusion/benz.jpg'						)

						lowerCAmelCase__		  =  torch.manual_seed(0						)
						lowerCAmelCase__		  =  pipe.dual_guided(
						    prompt='first prompt'    ,image=a_    ,text_to_image_strength=0.75    ,generator=a_    ,guidance_scale=7.5    ,num_inference_steps=2    ,output_type='numpy'    ,).images

						with tempfile.TemporaryDirectory() as tmpdirname:
										pipe.save_pretrained(a_						)
										lowerCAmelCase__		  =  VersatileDiffusionPipeline.from_pretrained(a_    ,torch_dtype=torch.floataa						)
						pipe.to(a_						)
						pipe.set_progress_bar_config(disable=a_						)

						lowerCAmelCase__		  =  generator.manual_seed(0						)
						lowerCAmelCase__		  =  pipe.dual_guided(
						    prompt='first prompt'    ,image=a_    ,text_to_image_strength=0.75    ,generator=a_    ,guidance_scale=7.5    ,num_inference_steps=2    ,output_type='numpy'    ,).images

						assert np.abs(image - new_image						).sum() < 1e-5, "Models don't have the same forward pass"



		def 						SCREAMING_SNAKE_CASE_					(							self						):







						"""simple docstring"""
						lowerCAmelCase__		  =  VersatileDiffusionPipeline.from_pretrained('shi-labs/versatile-diffusion'    ,torch_dtype=torch.floataa						)
						pipe.to(a_						)
						pipe.set_progress_bar_config(disable=a_						)

						lowerCAmelCase__		  =  'cyberpunk 2077'
						lowerCAmelCase__		  =  load_image(
						    'https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/versatile_diffusion/benz.jpg'						)
						lowerCAmelCase__		  =  torch.manual_seed(0						)
						lowerCAmelCase__		  =  pipe.dual_guided(
						    prompt=a_    ,image=a_    ,text_to_image_strength=0.75    ,generator=a_    ,guidance_scale=7.5    ,num_inference_steps=50    ,output_type='numpy'    ,).images

						lowerCAmelCase__		  =  image[0, 253:256, 253:256, -1]

						assert image.shape == (1, 512, 512, 3)
						lowerCAmelCase__		  =  np.array([0.1448, 0.1619, 0.1741, 0.1086, 0.1147, 0.1128, 0.1199, 0.1165, 0.1001]						)

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

						lowerCAmelCase__		  =  'A painting of a squirrel eating a burger '
						lowerCAmelCase__		  =  torch.manual_seed(0						)
						lowerCAmelCase__		  =  pipe.text_to_image(
						    prompt=a_    ,generator=a_    ,guidance_scale=7.5    ,num_inference_steps=50    ,output_type='numpy'						).images

						lowerCAmelCase__		  =  image[0, 253:256, 253:256, -1]

						assert image.shape == (1, 512, 512, 3)
						lowerCAmelCase__		  =  np.array([0.3367, 0.3169, 0.2656, 0.3870, 0.4790, 0.3796, 0.4009, 0.4878, 0.4778]						)

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

						lowerCAmelCase__		  =  pipe.image_variation(a_    ,generator=a_    ,output_type='numpy'						).images

						lowerCAmelCase__		  =  image[0, 253:256, 253:256, -1]

						assert image.shape == (1, 512, 512, 3)
						lowerCAmelCase__		  =  np.array([0.3076, 0.3123, 0.3284, 0.3782, 0.3770, 0.3894, 0.4297, 0.4331, 0.4456]						)

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