



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


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

SCREAMING_SNAKE_CASE_				= '''
    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_						(							SCREAMING_SNAKE_CASE__, SCREAMING_SNAKE_CASE__, SCREAMING_SNAKE_CASE__=8	)			->      Dict:
		a_     :      str 						=     height // scale_factor**2
		if height % scale_factor**2 != 0:
				new_height += 1
		a_     :      List[Any] 						=     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_       (						__UpperCAmelCase				):






							def __init__(   self			,     a_			,     a_			,     a_			,     ):
									super().__init__()

									self.register_modules(
									    unet=__SCREAMING_SNAKE_CASE			,     scheduler=__SCREAMING_SNAKE_CASE			,     movq=__SCREAMING_SNAKE_CASE			,     )
									a_     :      List[str] 						=     2 ** (len(self.movq.config.block_out_channels      ) - 1)






							def snake_case_  (   self			,     a_			,     a_			,     a_			,     a_			,     a_			,     a_      ):

									if latents is None:
											a_     :      Dict 						=     randn_tensor(__SCREAMING_SNAKE_CASE			,     generator=__SCREAMING_SNAKE_CASE			,     device=__SCREAMING_SNAKE_CASE			,     dtype=__SCREAMING_SNAKE_CASE      )
									else:
											if latents.shape != shape:
													raise ValueError(F"""Unexpected latents shape, got {latents.shape}, expected {shape}"""      )
											a_     :      Union[str, Any] 						=     latents.to(__SCREAMING_SNAKE_CASE      )

									a_     :      Tuple 						=     latents * scheduler.init_noise_sigma
									return latents






							def snake_case_  (   self			,     a_=0      ):

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

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

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






							def snake_case_  (   self			,     a_=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=__SCREAMING_SNAKE_CASE      )
											torch.cuda.empty_cache()  # otherwise we don't see the memory savings (but they probably exist)

									a_     :      List[Any] 						=     None
									for cpu_offloaded_model in [self.unet, self.movq]:
											a_						,				a_     :      Optional[int] 						=     cpu_offload_with_hook(__SCREAMING_SNAKE_CASE			,     __SCREAMING_SNAKE_CASE			,     prev_module_hook=__SCREAMING_SNAKE_CASE      )

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






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

									if not hasattr(self.unet			,     "_hf_hook"      ):
											return self.device
									for module in self.unet.modules():
											if (
											    hasattr(__SCREAMING_SNAKE_CASE			,     "_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(__SCREAMING_SNAKE_CASE      )
							def __call__(   self			,     a_			,     a_			,     a_ = 5_1_2			,     a_ = 5_1_2			,     a_ = 1_0_0			,     a_ = 4.0			,     a_ = 1			,     a_ = None			,     a_ = None			,     a_ = "pil"			,     a_ = True			,     ):
									a_     :      str 						=     self._execution_device

									a_     :      List[Any] 						=     guidance_scale > 1.0

									if isinstance(__SCREAMING_SNAKE_CASE			,     __SCREAMING_SNAKE_CASE      ):
											a_     :      Tuple 						=     torch.cat(__SCREAMING_SNAKE_CASE			,     dim=0      )
									a_     :      Optional[int] 						=     image_embeds.shape[0] * num_images_per_prompt
									if isinstance(__SCREAMING_SNAKE_CASE			,     __SCREAMING_SNAKE_CASE      ):
											a_     :      List[str] 						=     torch.cat(__SCREAMING_SNAKE_CASE			,     dim=0      )

									if do_classifier_free_guidance:
											a_     :      Any 						=     image_embeds.repeat_interleave(__SCREAMING_SNAKE_CASE			,     dim=0      )
											a_     :      Optional[int] 						=     negative_image_embeds.repeat_interleave(__SCREAMING_SNAKE_CASE			,     dim=0      )

									a_     :      Optional[Any] 						=     torch.cat([negative_image_embeds, image_embeds]			,     dim=0      ).to(dtype=self.unet.dtype			,     device=__SCREAMING_SNAKE_CASE      )

									self.scheduler.set_timesteps(__SCREAMING_SNAKE_CASE			,     device=__SCREAMING_SNAKE_CASE      )
									a_     :      str 						=     self.scheduler.timesteps

									a_     :      Dict 						=     self.unet.config.in_channels

									a_						,				a_     :      List[str] 						=     downscale_height_and_width(__SCREAMING_SNAKE_CASE			,     __SCREAMING_SNAKE_CASE			,     self.movq_scale_factor      )

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

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

											a_     :      Dict 						=     {"image_embeds": image_embeds}
											a_     :      Optional[int] 						=     self.unet(
											    sample=__SCREAMING_SNAKE_CASE			,     timestep=__SCREAMING_SNAKE_CASE			,     encoder_hidden_states=__SCREAMING_SNAKE_CASE			,     added_cond_kwargs=__SCREAMING_SNAKE_CASE			,     return_dict=__SCREAMING_SNAKE_CASE			,     )[0]

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

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

									if output_type == "pil":
											a_     :      Optional[int] 						=     self.numpy_to_pil(__SCREAMING_SNAKE_CASE      )

									if not return_dict:
											return (image,)

									return ImagePipelineOutput(images=__SCREAMING_SNAKE_CASE      )




"""simple docstring"""
from typing import Optional, Tuple, Union

import flax
import flax.linen as nn
import jax
import jax.numpy as jnp
from flax.core.frozen_dict import FrozenDict

from ..configuration_utils import ConfigMixin, flax_register_to_config
from ..utils import BaseOutput
from .embeddings_flax import FlaxTimestepEmbedding, FlaxTimesteps
from .modeling_flax_utils import FlaxModelMixin
from .unet_ad_blocks_flax import (
    FlaxCrossAttnDownBlockaD,
    FlaxDownBlockaD,
    FlaxUNetMidBlockaDCrossAttn,
)




@flax.struct.dataclass
class 			snake_case_						(						a_				):
							__lowerCAmelCase    					=					42
							__lowerCAmelCase    					=					42




class 			snake_case_						(						nn.Module				):
							__lowerCAmelCase    					=					42
							__lowerCAmelCase    					=					(1_6, 3_2, 9_6, 2_5_6)
							__lowerCAmelCase    					=					jnp.floataa






							def snake_case_  (   self      ):
									a_     :      Any 						=     nn.Conv(
									    self.block_out_channels[0]			,     kernel_size=(3, 3)			,     padding=((1, 1), (1, 1))			,     dtype=self.dtype			,     )

									a_     :      Tuple 						=     []
									for i in range(len(self.block_out_channels      ) - 1      ):
											a_     :      Tuple 						=     self.block_out_channels[i]
											a_     :      Any 						=     self.block_out_channels[i + 1]
											a_     :      Dict 						=     nn.Conv(
											    a_			,     kernel_size=(3, 3)			,     padding=((1, 1), (1, 1))			,     dtype=self.dtype			,     )
											blocks.append(a_      )
											a_     :      Optional[int] 						=     nn.Conv(
											    a_			,     kernel_size=(3, 3)			,     strides=(2, 2)			,     padding=((1, 1), (1, 1))			,     dtype=self.dtype			,     )
											blocks.append(a_      )
									a_     :      int 						=     blocks

									a_     :      str 						=     nn.Conv(
									    self.conditioning_embedding_channels			,     kernel_size=(3, 3)			,     padding=((1, 1), (1, 1))			,     kernel_init=nn.initializers.zeros_init()			,     bias_init=nn.initializers.zeros_init()			,     dtype=self.dtype			,     )






							def __call__(   self			,     a_      ):
									a_     :      Dict 						=     self.conv_in(a_      )
									a_     :      Optional[Any] 						=     nn.silu(a_      )

									for block in self.blocks:
											a_     :      Tuple 						=     block(a_      )
											a_     :      Union[str, Any] 						=     nn.silu(a_      )

									a_     :      List[str] 						=     self.conv_out(a_      )

									return embedding




@flax_register_to_config
class 			snake_case_						(						nn.Module     ,a_     ,a_				):
							__lowerCAmelCase    					=					3_2
							__lowerCAmelCase    					=					4
							__lowerCAmelCase    					=					(
							    "CrossAttnDownBlock2D",
							    "CrossAttnDownBlock2D",
							    "CrossAttnDownBlock2D",
							    "DownBlock2D",
							)
							__lowerCAmelCase    					=					False
							__lowerCAmelCase    					=					(3_2_0, 6_4_0, 1_2_8_0, 1_2_8_0)
							__lowerCAmelCase    					=					2
							__lowerCAmelCase    					=					8
							__lowerCAmelCase    					=					None
							__lowerCAmelCase    					=					1_2_8_0
							__lowerCAmelCase    					=					0.0
							__lowerCAmelCase    					=					False
							__lowerCAmelCase    					=					jnp.floataa
							__lowerCAmelCase    					=					True
							__lowerCAmelCase    					=					0
							__lowerCAmelCase    					=					"rgb"
							__lowerCAmelCase    					=					(1_6, 3_2, 9_6, 2_5_6)






							def snake_case_  (   self			,     a_      ):
									# init input tensors
									a_     :      Any 						=     (1, self.in_channels, self.sample_size, self.sample_size)
									a_     :      Tuple 						=     jnp.zeros(a_			,     dtype=jnp.floataa      )
									a_     :      Optional[int] 						=     jnp.ones((1,)			,     dtype=jnp.intaa      )
									a_     :      Any 						=     jnp.zeros((1, 1, self.cross_attention_dim)			,     dtype=jnp.floataa      )
									a_     :      Union[str, Any] 						=     (1, 3, self.sample_size * 8, self.sample_size * 8)
									a_     :      Optional[int] 						=     jnp.zeros(a_			,     dtype=jnp.floataa      )

									a_						,				a_     :      List[Any] 						=     jax.random.split(a_      )
									a_     :      Dict 						=     {"params": params_rng, "dropout": dropout_rng}

									return self.init(a_			,     a_			,     a_			,     a_			,     a_      )["params"]






							def snake_case_  (   self      ):
									a_     :      List[Any] 						=     self.block_out_channels
									a_     :      Any 						=     block_out_channels[0] * 4

									# If `num_attention_heads` is not defined (which is the case for most models)
									# it will default to `attention_head_dim`. This looks weird upon first reading it and it is.
									# The reason for this behavior is to correct for incorrectly named variables that were introduced
									# when this library was created. The incorrect naming was only discovered much later in https://github.com/huggingface/diffusers/issues/2011#issuecomment-1547958131
									# Changing `attention_head_dim` to `num_attention_heads` for 40,000+ configurations is too backwards breaking
									# which is why we correct for the naming here.
									a_     :      Optional[int] 						=     self.num_attention_heads or self.attention_head_dim

									# input
									a_     :      Optional[int] 						=     nn.Conv(
									    block_out_channels[0]			,     kernel_size=(3, 3)			,     strides=(1, 1)			,     padding=((1, 1), (1, 1))			,     dtype=self.dtype			,     )

									# time
									a_     :      Optional[Any] 						=     FlaxTimesteps(
									    block_out_channels[0]			,     flip_sin_to_cos=self.flip_sin_to_cos			,     freq_shift=self.config.freq_shift      )
									a_     :      List[str] 						=     FlaxTimestepEmbedding(a_			,     dtype=self.dtype      )

									a_     :      Optional[int] 						=     FlaxControlNetConditioningEmbedding(
									    conditioning_embedding_channels=block_out_channels[0]			,     block_out_channels=self.conditioning_embedding_out_channels			,     )

									a_     :      int 						=     self.only_cross_attention
									if isinstance(a_			,     a_      ):
											a_     :      Tuple 						=     (only_cross_attention,) * len(self.down_block_types      )

									if isinstance(a_			,     a_      ):
											a_     :      int 						=     (num_attention_heads,) * len(self.down_block_types      )

									# down
									a_     :      Any 						=     []
									a_     :      Optional[Any] 						=     []

									a_     :      List[Any] 						=     block_out_channels[0]

									a_     :      Dict 						=     nn.Conv(
									    a_			,     kernel_size=(1, 1)			,     padding="VALID"			,     kernel_init=nn.initializers.zeros_init()			,     bias_init=nn.initializers.zeros_init()			,     dtype=self.dtype			,     )
									controlnet_down_blocks.append(a_      )

									for i, down_block_type in enumerate(self.down_block_types      ):
											a_     :      int 						=     output_channel
											a_     :      Optional[int] 						=     block_out_channels[i]
											a_     :      str 						=     i == len(a_      ) - 1

											if down_block_type == "CrossAttnDownBlock2D":
													a_     :      int 						=     FlaxCrossAttnDownBlockaD(
													    in_channels=a_			,     out_channels=a_			,     dropout=self.dropout			,     num_layers=self.layers_per_block			,     num_attention_heads=num_attention_heads[i]			,     add_downsample=not is_final_block			,     use_linear_projection=self.use_linear_projection			,     only_cross_attention=only_cross_attention[i]			,     dtype=self.dtype			,     )
											else:
													a_     :      Union[str, Any] 						=     FlaxDownBlockaD(
													    in_channels=a_			,     out_channels=a_			,     dropout=self.dropout			,     num_layers=self.layers_per_block			,     add_downsample=not is_final_block			,     dtype=self.dtype			,     )

											down_blocks.append(a_      )

											for _ in range(self.layers_per_block      ):
													a_     :      Tuple 						=     nn.Conv(
													    a_			,     kernel_size=(1, 1)			,     padding="VALID"			,     kernel_init=nn.initializers.zeros_init()			,     bias_init=nn.initializers.zeros_init()			,     dtype=self.dtype			,     )
													controlnet_down_blocks.append(a_      )

											if not is_final_block:
													a_     :      Optional[Any] 						=     nn.Conv(
													    a_			,     kernel_size=(1, 1)			,     padding="VALID"			,     kernel_init=nn.initializers.zeros_init()			,     bias_init=nn.initializers.zeros_init()			,     dtype=self.dtype			,     )
													controlnet_down_blocks.append(a_      )

									a_     :      Union[str, Any] 						=     down_blocks
									a_     :      Optional[int] 						=     controlnet_down_blocks

									# mid
									a_     :      Union[str, Any] 						=     block_out_channels[-1]
									a_     :      Any 						=     FlaxUNetMidBlockaDCrossAttn(
									    in_channels=a_			,     dropout=self.dropout			,     num_attention_heads=num_attention_heads[-1]			,     use_linear_projection=self.use_linear_projection			,     dtype=self.dtype			,     )

									a_     :      Union[str, Any] 						=     nn.Conv(
									    a_			,     kernel_size=(1, 1)			,     padding="VALID"			,     kernel_init=nn.initializers.zeros_init()			,     bias_init=nn.initializers.zeros_init()			,     dtype=self.dtype			,     )






							def __call__(   self			,     a_			,     a_			,     a_			,     a_			,     a_ = 1.0			,     a_ = True			,     a_ = False			,     ):
									a_     :      int 						=     self.controlnet_conditioning_channel_order
									if channel_order == "bgr":
											a_     :      Tuple 						=     jnp.flip(a_			,     axis=1      )

									# 1. time
									if not isinstance(a_			,     jnp.ndarray      ):
											a_     :      Optional[int] 						=     jnp.array([timesteps]			,     dtype=jnp.intaa      )
									elif isinstance(a_			,     jnp.ndarray      ) and len(timesteps.shape      ) == 0:
											a_     :      Any 						=     timesteps.astype(dtype=jnp.floataa      )
											a_     :      List[Any] 						=     jnp.expand_dims(a_			,     0      )

									a_     :      List[str] 						=     self.time_proj(a_      )
									a_     :      Union[str, Any] 						=     self.time_embedding(a_      )

									# 2. pre-process
									a_     :      Union[str, Any] 						=     jnp.transpose(a_			,     (0, 2, 3, 1)      )
									a_     :      Union[str, Any] 						=     self.conv_in(a_      )

									a_     :      Optional[int] 						=     jnp.transpose(a_			,     (0, 2, 3, 1)      )
									a_     :      int 						=     self.controlnet_cond_embedding(a_      )
									sample += controlnet_cond

									# 3. down
									a_     :      List[str] 						=     (sample,)
									for down_block in self.down_blocks:
											if isinstance(a_			,     a_      ):
													a_						,				a_     :      int 						=     down_block(a_			,     a_			,     a_			,     deterministic=not train      )
											else:
													a_						,				a_     :      Union[str, Any] 						=     down_block(a_			,     a_			,     deterministic=not train      )
											down_block_res_samples += res_samples

									# 4. mid
									a_     :      Optional[Any] 						=     self.mid_block(a_			,     a_			,     a_			,     deterministic=not train      )

									# 5. contronet blocks
									a_     :      Optional[int] 						=     ()
									for down_block_res_sample, controlnet_block in zip(a_			,     self.controlnet_down_blocks      ):
											a_     :      Tuple 						=     controlnet_block(a_      )
											controlnet_down_block_res_samples += (down_block_res_sample,)

									a_     :      Union[str, Any] 						=     controlnet_down_block_res_samples

									a_     :      Optional[int] 						=     self.controlnet_mid_block(a_      )

									# 6. scaling
									a_     :      Optional[Any] 						=     [sample * conditioning_scale for sample in down_block_res_samples]
									mid_block_res_sample *= conditioning_scale

									if not return_dict:
											return (down_block_res_samples, mid_block_res_sample)

									return FlaxControlNetOutput(
									    down_block_res_samples=a_			,     mid_block_res_sample=a_      )