






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

a__       =     """
    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   (	a						:	List[str]  ,					a						:	Tuple  ,					a						:	int=8 ):
							snake_case__	  =       height // scale_factor**2
							if height % scale_factor**2 != 0:
														new_height += 1
							snake_case__	  =       width // scale_factor**2
							if width % scale_factor**2 != 0:
														new_width += 1
							return new_height * scale_factor, new_width * scale_factor



class    _lowerCAmelCase			(		lowercase_      ):



				"""simple docstring"""



				def __init__(      self    :    Any			,		UpperCamelCase__    :    UNetaDConditionModel			,		UpperCamelCase__    :    DDPMScheduler			,		UpperCamelCase__    :    VQModel			,		):






											'''simple docstring'''







											super().__init__()

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



				def 					__magic_name__							(      self    :    Any			,		UpperCamelCase__    :    Tuple			,		UpperCamelCase__    :    Optional[int]			,		UpperCamelCase__    :    Tuple			,		UpperCamelCase__    :    List[Any]			,		UpperCamelCase__    :    List[str]			,		UpperCamelCase__    :    Any):






											'''simple docstring'''







											if latents is None:
																		snake_case__	  =       randn_tensor(UpperCamelCase__			,		generator=UpperCamelCase__			,		device=UpperCamelCase__			,		dtype=UpperCamelCase__)
											else:
																		if latents.shape != shape:
																									raise ValueError(F'''Unexpected latents shape, got {latents.shape}, expected {shape}''')
																		snake_case__	  =       latents.to(UpperCamelCase__)

											snake_case__	  =       latents * scheduler.init_noise_sigma
											return latents



				def 					__magic_name__							(      self    :    Union[str, Any]			,		UpperCamelCase__    :    Optional[int]=0):






											'''simple docstring'''







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

											snake_case__	  =       torch.device(F'''cuda:{gpu_id}''')

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



				def 					__magic_name__							(      self    :    List[str]			,		UpperCamelCase__    :    str=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.""")

											snake_case__	  =       torch.device(F'''cuda:{gpu_id}''')

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

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

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



				@property
				# Copied from diffusers.pipelines.stable_diffusion.pipeline_stable_diffusion.StableDiffusionPipeline._execution_device
				def 					__magic_name__							(      self    :    int):






											'''simple docstring'''







											if not hasattr(self.unet			,		"""_hf_hook"""):
																		return self.device
											for module in self.unet.modules():
																		if (
																		    hasattr(UpperCamelCase__			,		"""_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(UpperCamelCase__)
				def __call__(      self    :    Tuple			,		UpperCamelCase__    :    Union[torch.FloatTensor, List[torch.FloatTensor]]			,		UpperCamelCase__    :    Union[torch.FloatTensor, List[torch.FloatTensor]]			,		UpperCamelCase__    :    int = 5_1_2			,		UpperCamelCase__    :    int = 5_1_2			,		UpperCamelCase__    :    int = 1_0_0			,		UpperCamelCase__    :    float = 4.0			,		UpperCamelCase__    :    int = 1			,		UpperCamelCase__    :    Optional[Union[torch.Generator, List[torch.Generator]]] = None			,		UpperCamelCase__    :    Optional[torch.FloatTensor] = None			,		UpperCamelCase__    :    Optional[str] = "pil"			,		UpperCamelCase__    :    bool = True			,		):






											'''simple docstring'''







											snake_case__	  =       self._execution_device

											snake_case__	  =       guidance_scale > 1.0

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

											if do_classifier_free_guidance:
																		snake_case__	  =       image_embeds.repeat_interleave(UpperCamelCase__			,		dim=0)
																		snake_case__	  =       negative_image_embeds.repeat_interleave(UpperCamelCase__			,		dim=0)

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

											self.scheduler.set_timesteps(UpperCamelCase__			,		device=UpperCamelCase__)
											snake_case__	  =       self.scheduler.timesteps

											snake_case__	  =       self.unet.config.in_channels

											snake_case__  , snake_case__	  =       downscale_height_and_width(UpperCamelCase__			,		UpperCamelCase__			,		self.movq_scale_factor)

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

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

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

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

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

											if output_type == "pil":
																		snake_case__	  =       self.numpy_to_pil(UpperCamelCase__)

											if not return_dict:
																		return (image,)

											return ImagePipelineOutput(images=UpperCamelCase__)








from dataclasses import dataclass, field
from typing import Tuple

from ..utils import cached_property, is_torch_available, is_torch_tpu_available, logging, requires_backends
from .benchmark_args_utils import BenchmarkArguments


if is_torch_available():
	import torch

if is_torch_tpu_available(check_device=False):
	import torch_xla.core.xla_model as xm


a__       =     logging.get_logger(__name__)


@dataclass
class    _lowerCAmelCase			(		lowercase_      ):



				"""simple docstring"""



				_lowercase				:   Dict      =  [
				    '''no_inference''',
				    '''no_cuda''',
				    '''no_tpu''',
				    '''no_speed''',
				    '''no_memory''',
				    '''no_env_print''',
				    '''no_multi_process''',
				]



				def __init__(      self    :    str			,		**UpperCamelCase__    :    Dict):






											'''simple docstring'''







											for deprecated_arg in self.deprecated_args:
																		if deprecated_arg in kwargs:
																									snake_case__	  =       deprecated_arg[3:]
																									setattr(self			,		UpperCamelCase__			,		not kwargs.pop(UpperCamelCase__))
																									logger.warning(
																									    F'''{deprecated_arg} is depreciated. Please use --no_{positive_arg} or'''
																									    F''' {positive_arg}={kwargs[positive_arg]}''')

											snake_case__	  =       kwargs.pop("""torchscript"""			,		self.torchscript)
											snake_case__	  =       kwargs.pop("""torch_xla_tpu_print_metrics"""			,		self.torch_xla_tpu_print_metrics)
											snake_case__	  =       kwargs.pop("""fp16_opt_level"""			,		self.fpaa_opt_level)
											super().__init__(**UpperCamelCase__)

				_lowercase				:   bool      =  field(default=lowercase_     ,      metadata={'''help''': '''Trace the models using torchscript'''}      )
				_lowercase				:   bool      =  field(default=lowercase_     ,      metadata={'''help''': '''Print Xla/PyTorch tpu metrics'''}      )
				_lowercase				:   str      =  field(
				    default='''O1'''     ,      metadata={
				        '''help''': (
				            '''For fp16: Apex AMP optimization level selected in [\'O0\', \'O1\', \'O2\', and \'O3\']. '''
				            '''See details at https://nvidia.github.io/apex/amp.html'''
				        )
				    }     ,      )



				@cached_property
				def 					__magic_name__							(      self    :    Tuple):






											'''simple docstring'''







											requires_backends(self			,		["""torch"""])
											logger.info("""PyTorch: setting up devices""")
											if not self.cuda:
																		snake_case__	  =       torch.device("""cpu""")
																		snake_case__	  =       0
											elif is_torch_tpu_available():
																		snake_case__	  =       xm.xla_device()
																		snake_case__	  =       0
											else:
																		snake_case__	  =       torch.device("""cuda""" if torch.cuda.is_available() else """cpu""")
																		snake_case__	  =       torch.cuda.device_count()
											return device, n_gpu



				@property
				def 					__magic_name__							(      self    :    Union[str, Any]):






											'''simple docstring'''







											return is_torch_tpu_available() and self.tpu



				@property
				def 					__magic_name__							(      self    :    List[str]):






											'''simple docstring'''







											requires_backends(self			,		["""torch"""])
											# TODO(PVP): currently only single GPU is supported
											return torch.cuda.current_device()



				@property
				def 					__magic_name__							(      self    :    Union[str, Any]):






											'''simple docstring'''







											requires_backends(self			,		["""torch"""])
											return self._setup_devices[0]



				@property
				def 					__magic_name__							(      self    :    str):






											'''simple docstring'''







											requires_backends(self			,		["""torch"""])
											return self._setup_devices[1]



				@property
				def 					__magic_name__							(      self    :    str):






											'''simple docstring'''







											return self.n_gpu > 0
