




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


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     ( __UpperCamelCase	,				__UpperCamelCase	,				__UpperCamelCase=8	):
			__lowercase			:  Optional[Any]								=       height // scale_factor**2
			if height % scale_factor**2 != 0:
						new_height += 1
			__lowercase			:  Tuple								=       width // scale_factor**2
			if width % scale_factor**2 != 0:
						new_width += 1
			return new_height * scale_factor, new_width * scale_factor


class   UpperCAmelCase_    (			__SCREAMING_SNAKE_CASE		):
		def __init__(						self				,						UpperCamelCase_				,						UpperCamelCase_				,						UpperCamelCase_				,						)						->       Union[str, Any]:
					super().__init__()

					self.register_modules(
					    unet=A_				,						scheduler=A_				,						movq=A_				,						)
					__lowercase			:  Any								=       2 ** (len(self.movq.config.block_out_channels       ) - 1)
		def 	_lowerCamelCase							(						self				,						UpperCamelCase_				,						UpperCamelCase_				,						UpperCamelCase_				,						UpperCamelCase_				,						UpperCamelCase_				,						UpperCamelCase_       )						->       Tuple:

					if latents is None:
								__lowercase			:  Optional[int]								=       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}"""       )
								__lowercase			:  Optional[int]								=       latents.to(A_       )

					__lowercase			:  Union[str, Any]								=       latents * scheduler.init_noise_sigma
					return latents
		def 	_lowerCamelCase							(						self				,						UpperCamelCase_=0       )						->       Any:

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

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

					__lowercase			:  Tuple								=       [
					    self.unet,
					    self.movq,
					]
					for cpu_offloaded_model in models:
								if cpu_offloaded_model is not None:
											cpu_offload(A_				,						A_       )
		def 	_lowerCamelCase							(						self				,						UpperCamelCase_=0       )						->       List[Any]:

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

					__lowercase			:  List[Any]								=       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)

					__lowercase			:  Optional[int]								=       None
					for cpu_offloaded_model in [self.unet, self.movq]:
								__lowercase			,__lowercase			:  str								=       cpu_offload_with_hook(A_				,						A_				,						prev_module_hook=A_       )

					# We'll offload the last model manually.
					__lowercase			:  int								=       hook
		@property
		# Copied from diffusers.pipelines.stable_diffusion.pipeline_stable_diffusion.StableDiffusionPipeline._execution_device
		def 	_lowerCamelCase							(						self       )						->       Optional[Any]:

					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				,						UpperCamelCase_				,						UpperCamelCase_				,						UpperCamelCase_ = 5_12				,						UpperCamelCase_ = 5_12				,						UpperCamelCase_ = 1_00				,						UpperCamelCase_ = 4.0				,						UpperCamelCase_ = 1				,						UpperCamelCase_ = None				,						UpperCamelCase_ = None				,						UpperCamelCase_ = "pil"				,						UpperCamelCase_ = True				,						)						->       Optional[int]:
					__lowercase			:  Any								=       self._execution_device

					__lowercase			:  Dict								=       guidance_scale > 1.0

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

					if do_classifier_free_guidance:
								__lowercase			:  Optional[int]								=       image_embeds.repeat_interleave(A_				,						dim=0       )
								__lowercase			:  Tuple								=       negative_image_embeds.repeat_interleave(A_				,						dim=0       )

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

					self.scheduler.set_timesteps(A_				,						device=A_       )
					__lowercase			:  Dict								=       self.scheduler.timesteps

					__lowercase			:  Union[str, Any]								=       self.unet.config.in_channels

					__lowercase			,__lowercase			:  Union[str, Any]								=       downscale_height_and_width(A_				,						A_				,						self.movq_scale_factor       )

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

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

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

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

					if output_type == "pil":
								__lowercase			:  Any								=       self.numpy_to_pil(A_       )

					if not return_dict:
								return (image,)

					return ImagePipelineOutput(images=A_       )










import re

from flax.core.frozen_dict import freeze
from flax.traverse_util import flatten_dict, unflatten_dict
from jax.experimental import PartitionSpec as P


# Sentinels
_A					:				Optional[int]				=			object()

# For specifying empty leaf dict `{}`
_A					:				Tuple				=			object()


def 							__snake_case							(    lowerCAmelCase_ ,							lowerCAmelCase_	)				->						Dict:
    SCREAMING_SNAKE_CASE__											=			tuple((re.compile(x + '''$'''	) for x in qs)	)
    for i in range(len(lowerCAmelCase_	) - len(lowerCAmelCase_	) + 1	):
        SCREAMING_SNAKE_CASE__											=			[x.match(lowerCAmelCase_	) for x, y in zip(lowerCAmelCase_ ,							ks[i:]	)]
        if matches and all(lowerCAmelCase_	):
            return True
    return False


def 							__snake_case							(    lowerCAmelCase_	)				->						int:
    def replace(lowerCAmelCase_ ,							lowerCAmelCase_	):
        for rule, replacement in rules:
            if _match(lowerCAmelCase_ ,							lowerCAmelCase_	):
                return replacement
        return val

    return replace


def 							__snake_case							(    )				->						Optional[int]:
    return [
        # embeddings
        (("transformer", "wpe", "embedding"), P('''mp''' ,							lowerCAmelCase_	)),
        (("transformer", "wte", "embedding"), P('''mp''' ,							lowerCAmelCase_	)),
        # atention
        (("attention", "(q_proj|k_proj|v_proj)", "kernel"), P(lowerCAmelCase_ ,							'''mp'''	)),
        (("attention", "out_proj", "kernel"), P('''mp''' ,							lowerCAmelCase_	)),
        (("attention", "out_proj", "bias"), None),
        # mlp
        (("mlp", "c_fc", "kernel"), P(lowerCAmelCase_ ,							'''mp'''	)),
        (("mlp", "c_fc", "bias"), P('''mp'''	)),
        (("mlp", "c_proj", "kernel"), P('''mp''' ,							lowerCAmelCase_	)),
        (("mlp", "c_proj", "bias"), None),
        # layer norms
        ((r"ln_\d+", "bias"), None),
        ((r"\d+", r"ln_\d+", "scale"), None),
        (("ln_f", "bias"), None),
        (("ln_f", "scale"), None),
    ]



def 							__snake_case							(    lowerCAmelCase_	)				->						Optional[Any]:
    SCREAMING_SNAKE_CASE__											=			_get_partition_rules()
    SCREAMING_SNAKE_CASE__											=			_replacement_rules(lowerCAmelCase_	)
    SCREAMING_SNAKE_CASE__											=			{k: _unmatched for k in flatten_dict(lowerCAmelCase_	)}
    SCREAMING_SNAKE_CASE__											=			{k: replace(lowerCAmelCase_ ,							lowerCAmelCase_	) for k, v in initd.items()}
    assert _unmatched not in result.values(), "Incomplete partition spec."
    return freeze(unflatten_dict(lowerCAmelCase_	)	)
