


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


_lowercase      :      Optional[int]            =					logging.get_logger(__name__)  # pylint: disable=invalid-name

_lowercase      :      str            =					"""
    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 				lowerCamelCase__				(	A       :      Any				,       A       :      Optional[Any]				,       A       :      Any=8					):




      '''simple docstring'''






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






class 	UpperCamelCase__(       lowerCAmelCase	):




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

          """simple docstring"""




          super().__init__()

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




    def 					a__(		self      :			int				,   lowerCAmelCase      :			Dict				,   lowerCAmelCase      :			List[Any]				,   lowerCAmelCase      :			int				,   lowerCAmelCase      :			Any				,   lowerCAmelCase      :			Tuple				,   lowerCAmelCase      :			str				)->					str:

          """simple docstring"""




          if latents is None:
                UpperCAmelCase					    =     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}"""				)
                UpperCAmelCase					    =     latents.to(lowerCAmelCase				)

          UpperCAmelCase					    =     latents * scheduler.init_noise_sigma
          return latents




    def 					a__(		self      :			Dict				,   lowerCAmelCase      :			Union[str, Any]=0				)->					Optional[Any]:

          """simple docstring"""




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

          UpperCAmelCase					    =     torch.device(F"""cuda:{gpu_id}"""				)

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




    def 					a__(		self      :			str				,   lowerCAmelCase      :			Union[str, Any]=0				)->					List[str]:

          """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.'''				)

          UpperCAmelCase					    =     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)

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

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




    @property
    # Copied from diffusers.pipelines.stable_diffusion.pipeline_stable_diffusion.StableDiffusionPipeline._execution_device
    def 					a__(		self      :			int				)->					Union[str, Any]:

          """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      :			Union[str, Any]				,   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				,   )->					int:

          """simple docstring"""




          UpperCAmelCase					    =     self._execution_device

          UpperCAmelCase					    =     guidance_scale > 1.0

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

          if do_classifier_free_guidance:
                UpperCAmelCase					    =     image_embeds.repeat_interleave(lowerCAmelCase				,   dim=0				)
                UpperCAmelCase					    =     negative_image_embeds.repeat_interleave(lowerCAmelCase				,   dim=0				)

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

          self.scheduler.set_timesteps(lowerCAmelCase				,   device=lowerCAmelCase				)
          UpperCAmelCase					    =     self.scheduler.timesteps

          UpperCAmelCase					    =     self.unet.config.in_channels

          UpperCAmelCase    ,    UpperCAmelCase					    =     downscale_height_and_width(lowerCAmelCase				,   lowerCAmelCase				,   self.movq_scale_factor				)

          # create initial latent
          UpperCAmelCase					    =     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
                UpperCAmelCase					    =     torch.cat([latents] * 2				) if do_classifier_free_guidance else latents

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

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

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

          if output_type == "pil":
                UpperCAmelCase					    =     self.numpy_to_pil(lowerCAmelCase				)

          if not return_dict:
                return (image,)

          return ImagePipelineOutput(images=lowerCAmelCase				)









'''simple docstring'''



import flax.linen as nn
import jax.numpy as jnp

from .attention_flax import FlaxTransformeraDModel
from .resnet_flax import FlaxDownsampleaD, FlaxResnetBlockaD, FlaxUpsampleaD






class 	UpperCamelCase__(       nn.Module	):
				__magic_name__    :				int
				__magic_name__    :				int
				__magic_name__    :				float  		=	0.0
				__magic_name__    :				int  		=	1
				__magic_name__    :				int  		=	1
				__magic_name__    :				bool  		=	True
				__magic_name__    :				bool  		=	False
				__magic_name__    :				bool  		=	False
				__magic_name__    :				bool  		=	False
				__magic_name__    :				jnp.dtype  		=	jnp.floataa




				def 					a__(		self      :			str				)->					Dict:

										"""simple docstring"""




										UpperCAmelCase					    =     []
										UpperCAmelCase					    =     []

										for i in range(self.num_layers				):
																UpperCAmelCase					    =     self.in_channels if i == 0 else self.out_channels

																UpperCAmelCase					    =     FlaxResnetBlockaD(
																    in_channels=lowerCAmelCase				,   out_channels=self.out_channels				,   dropout_prob=self.dropout				,   dtype=self.dtype				,   )
																resnets.append(lowerCAmelCase				)

																UpperCAmelCase					    =     FlaxTransformeraDModel(
																    in_channels=self.out_channels				,   n_heads=self.num_attention_heads				,   d_head=self.out_channels // self.num_attention_heads				,   depth=1				,   use_linear_projection=self.use_linear_projection				,   only_cross_attention=self.only_cross_attention				,   use_memory_efficient_attention=self.use_memory_efficient_attention				,   dtype=self.dtype				,   )
																attentions.append(lowerCAmelCase				)

										UpperCAmelCase					    =     resnets
										UpperCAmelCase					    =     attentions

										if self.add_downsample:
																UpperCAmelCase					    =     FlaxDownsampleaD(self.out_channels				,   dtype=self.dtype				)





				def __call__(		self      :			List[Any]				,   lowerCAmelCase      :			Any				,   lowerCAmelCase      :			int				,   lowerCAmelCase      :			Dict				,   lowerCAmelCase      :			List[str]=True				)->					Optional[int]:

										"""simple docstring"""




										UpperCAmelCase					    =     ()

										for resnet, attn in zip(self.resnets				,   self.attentions				):
																UpperCAmelCase					    =     resnet(lowerCAmelCase				,   lowerCAmelCase				,   deterministic=lowerCAmelCase				)
																UpperCAmelCase					    =     attn(lowerCAmelCase				,   lowerCAmelCase				,   deterministic=lowerCAmelCase				)
																output_states += (hidden_states,)

										if self.add_downsample:
																UpperCAmelCase					    =     self.downsamplers_a(lowerCAmelCase				)
																output_states += (hidden_states,)

										return hidden_states, output_states







class 	UpperCamelCase__(       nn.Module	):
				__magic_name__    :				int
				__magic_name__    :				int
				__magic_name__    :				float  		=	0.0
				__magic_name__    :				int  		=	1
				__magic_name__    :				bool  		=	True
				__magic_name__    :				jnp.dtype  		=	jnp.floataa




				def 					a__(		self      :			List[str]				)->					Any:

										"""simple docstring"""




										UpperCAmelCase					    =     []

										for i in range(self.num_layers				):
																UpperCAmelCase					    =     self.in_channels if i == 0 else self.out_channels

																UpperCAmelCase					    =     FlaxResnetBlockaD(
																    in_channels=lowerCAmelCase				,   out_channels=self.out_channels				,   dropout_prob=self.dropout				,   dtype=self.dtype				,   )
																resnets.append(lowerCAmelCase				)
										UpperCAmelCase					    =     resnets

										if self.add_downsample:
																UpperCAmelCase					    =     FlaxDownsampleaD(self.out_channels				,   dtype=self.dtype				)





				def __call__(		self      :			List[str]				,   lowerCAmelCase      :			Tuple				,   lowerCAmelCase      :			Dict				,   lowerCAmelCase      :			List[str]=True				)->					Optional[int]:

										"""simple docstring"""




										UpperCAmelCase					    =     ()

										for resnet in self.resnets:
																UpperCAmelCase					    =     resnet(lowerCAmelCase				,   lowerCAmelCase				,   deterministic=lowerCAmelCase				)
																output_states += (hidden_states,)

										if self.add_downsample:
																UpperCAmelCase					    =     self.downsamplers_a(lowerCAmelCase				)
																output_states += (hidden_states,)

										return hidden_states, output_states







class 	UpperCamelCase__(       nn.Module	):
				__magic_name__    :				int
				__magic_name__    :				int
				__magic_name__    :				int
				__magic_name__    :				float  		=	0.0
				__magic_name__    :				int  		=	1
				__magic_name__    :				int  		=	1
				__magic_name__    :				bool  		=	True
				__magic_name__    :				bool  		=	False
				__magic_name__    :				bool  		=	False
				__magic_name__    :				bool  		=	False
				__magic_name__    :				jnp.dtype  		=	jnp.floataa




				def 					a__(		self      :			List[str]				)->					Tuple:

										"""simple docstring"""




										UpperCAmelCase					    =     []
										UpperCAmelCase					    =     []

										for i in range(self.num_layers				):
																UpperCAmelCase					    =     self.in_channels if (i == self.num_layers - 1) else self.out_channels
																UpperCAmelCase					    =     self.prev_output_channel if i == 0 else self.out_channels

																UpperCAmelCase					    =     FlaxResnetBlockaD(
																    in_channels=resnet_in_channels + res_skip_channels				,   out_channels=self.out_channels				,   dropout_prob=self.dropout				,   dtype=self.dtype				,   )
																resnets.append(lowerCAmelCase				)

																UpperCAmelCase					    =     FlaxTransformeraDModel(
																    in_channels=self.out_channels				,   n_heads=self.num_attention_heads				,   d_head=self.out_channels // self.num_attention_heads				,   depth=1				,   use_linear_projection=self.use_linear_projection				,   only_cross_attention=self.only_cross_attention				,   use_memory_efficient_attention=self.use_memory_efficient_attention				,   dtype=self.dtype				,   )
																attentions.append(lowerCAmelCase				)

										UpperCAmelCase					    =     resnets
										UpperCAmelCase					    =     attentions

										if self.add_upsample:
																UpperCAmelCase					    =     FlaxUpsampleaD(self.out_channels				,   dtype=self.dtype				)





				def __call__(		self      :			Union[str, Any]				,   lowerCAmelCase      :			List[str]				,   lowerCAmelCase      :			Optional[int]				,   lowerCAmelCase      :			List[str]				,   lowerCAmelCase      :			Any				,   lowerCAmelCase      :			Union[str, Any]=True				)->					Optional[int]:

										"""simple docstring"""




										for resnet, attn in zip(self.resnets				,   self.attentions				):
																# pop res hidden states
																UpperCAmelCase					    =     res_hidden_states_tuple[-1]
																UpperCAmelCase					    =     res_hidden_states_tuple[:-1]
																UpperCAmelCase					    =     jnp.concatenate((hidden_states, res_hidden_states)				,   axis=-1				)

																UpperCAmelCase					    =     resnet(lowerCAmelCase				,   lowerCAmelCase				,   deterministic=lowerCAmelCase				)
																UpperCAmelCase					    =     attn(lowerCAmelCase				,   lowerCAmelCase				,   deterministic=lowerCAmelCase				)

										if self.add_upsample:
																UpperCAmelCase					    =     self.upsamplers_a(lowerCAmelCase				)

										return hidden_states







class 	UpperCamelCase__(       nn.Module	):
				__magic_name__    :				int
				__magic_name__    :				int
				__magic_name__    :				int
				__magic_name__    :				float  		=	0.0
				__magic_name__    :				int  		=	1
				__magic_name__    :				bool  		=	True
				__magic_name__    :				jnp.dtype  		=	jnp.floataa




				def 					a__(		self      :			Optional[int]				)->					str:

										"""simple docstring"""




										UpperCAmelCase					    =     []

										for i in range(self.num_layers				):
																UpperCAmelCase					    =     self.in_channels if (i == self.num_layers - 1) else self.out_channels
																UpperCAmelCase					    =     self.prev_output_channel if i == 0 else self.out_channels

																UpperCAmelCase					    =     FlaxResnetBlockaD(
																    in_channels=resnet_in_channels + res_skip_channels				,   out_channels=self.out_channels				,   dropout_prob=self.dropout				,   dtype=self.dtype				,   )
																resnets.append(lowerCAmelCase				)

										UpperCAmelCase					    =     resnets

										if self.add_upsample:
																UpperCAmelCase					    =     FlaxUpsampleaD(self.out_channels				,   dtype=self.dtype				)





				def __call__(		self      :			Union[str, Any]				,   lowerCAmelCase      :			Dict				,   lowerCAmelCase      :			Union[str, Any]				,   lowerCAmelCase      :			Tuple				,   lowerCAmelCase      :			Dict=True				)->					Tuple:

										"""simple docstring"""




										for resnet in self.resnets:
																# pop res hidden states
																UpperCAmelCase					    =     res_hidden_states_tuple[-1]
																UpperCAmelCase					    =     res_hidden_states_tuple[:-1]
																UpperCAmelCase					    =     jnp.concatenate((hidden_states, res_hidden_states)				,   axis=-1				)

																UpperCAmelCase					    =     resnet(lowerCAmelCase				,   lowerCAmelCase				,   deterministic=lowerCAmelCase				)

										if self.add_upsample:
																UpperCAmelCase					    =     self.upsamplers_a(lowerCAmelCase				)

										return hidden_states







class 	UpperCamelCase__(       nn.Module	):
				__magic_name__    :				int
				__magic_name__    :				float  		=	0.0
				__magic_name__    :				int  		=	1
				__magic_name__    :				int  		=	1
				__magic_name__    :				bool  		=	False
				__magic_name__    :				bool  		=	False
				__magic_name__    :				jnp.dtype  		=	jnp.floataa




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

										"""simple docstring"""




										UpperCAmelCase					    =     [
										    FlaxResnetBlockaD(
										        in_channels=self.in_channels				,   out_channels=self.in_channels				,   dropout_prob=self.dropout				,   dtype=self.dtype				,   )
										]

										UpperCAmelCase					    =     []

										for _ in range(self.num_layers				):
																UpperCAmelCase					    =     FlaxTransformeraDModel(
																    in_channels=self.in_channels				,   n_heads=self.num_attention_heads				,   d_head=self.in_channels // self.num_attention_heads				,   depth=1				,   use_linear_projection=self.use_linear_projection				,   use_memory_efficient_attention=self.use_memory_efficient_attention				,   dtype=self.dtype				,   )
																attentions.append(lowerCAmelCase				)

																UpperCAmelCase					    =     FlaxResnetBlockaD(
																    in_channels=self.in_channels				,   out_channels=self.in_channels				,   dropout_prob=self.dropout				,   dtype=self.dtype				,   )
																resnets.append(lowerCAmelCase				)

										UpperCAmelCase					    =     resnets
										UpperCAmelCase					    =     attentions





				def __call__(		self      :			Dict				,   lowerCAmelCase      :			Optional[Any]				,   lowerCAmelCase      :			Tuple				,   lowerCAmelCase      :			Dict				,   lowerCAmelCase      :			Any=True				)->					Union[str, Any]:

										"""simple docstring"""




										UpperCAmelCase					    =     self.resnets[0](lowerCAmelCase				,   lowerCAmelCase				)
										for attn, resnet in zip(self.attentions				,   self.resnets[1:]				):
																UpperCAmelCase					    =     attn(lowerCAmelCase				,   lowerCAmelCase				,   deterministic=lowerCAmelCase				)
																UpperCAmelCase					    =     resnet(lowerCAmelCase				,   lowerCAmelCase				,   deterministic=lowerCAmelCase				)

										return hidden_states





