


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


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

lowerCAmelCase__				    =       '''
    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  a__      (    SCREAMING_SNAKE_CASE   :   Dict    ,				SCREAMING_SNAKE_CASE   :   Optional[Any]    ,				SCREAMING_SNAKE_CASE   :   str=8    ):


		'''simple docstring'''







		lowerCAmelCase					:					Union[str, Any]								=							height // scale_factor**2
		if height % scale_factor**2 != 0:
				new_height += 1
		lowerCAmelCase					:					Tuple								=							width // scale_factor**2
		if width % scale_factor**2 != 0:
				new_width += 1
		return new_height * scale_factor, new_width * scale_factor





class 				SCREAMING_SNAKE_CASE__    (		_snake_case    ):
							"""simple docstring"""


							def __init__(						self		,					snake_case__		,					snake_case__		,					snake_case__		,					):






									"""simple docstring"""





									super().__init__()

									self.register_modules(
									    unet=UpperCamelCase__		,					scheduler=UpperCamelCase__		,					movq=UpperCamelCase__		,					)
									lowerCAmelCase					:					Optional[Any]								=							2 ** (len(self.movq.config.block_out_channels						) - 1)

							def 	lowercase__							(						self		,					snake_case__		,					snake_case__		,					snake_case__		,					snake_case__		,					snake_case__		,					snake_case__						):






									"""simple docstring"""





									if latents is None:
											lowerCAmelCase					:					List[str]								=							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}"""						)
											lowerCAmelCase					:					str								=							latents.to(UpperCamelCase__						)

									lowerCAmelCase					:					Tuple								=							latents * scheduler.init_noise_sigma
									return latents

							def 	lowercase__							(						self		,					snake_case__=0						):






									"""simple docstring"""





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

									lowerCAmelCase					:					Tuple								=							torch.device(f"""cuda:{gpu_id}"""						)

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

							def 	lowercase__							(						self		,					snake_case__=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					:					Any								=							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)

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

									# We'll offload the last model manually.
									lowerCAmelCase					:					str								=							hook

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






									"""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		,					snake_case__		,					snake_case__		,					snake_case__ = 512		,					snake_case__ = 512		,					snake_case__ = 100		,					snake_case__ = 4.0		,					snake_case__ = 1		,					snake_case__ = None		,					snake_case__ = None		,					snake_case__ = "pil"		,					snake_case__ = True		,					):






									"""simple docstring"""





									lowerCAmelCase					:					str								=							self._execution_device

									lowerCAmelCase					:					Dict								=							guidance_scale > 1.0

									if isinstance(UpperCamelCase__		,					UpperCamelCase__						):
											lowerCAmelCase					:					Optional[int]								=							torch.cat(UpperCamelCase__		,					dim=0						)
									lowerCAmelCase					:					Dict								=							image_embeds.shape[0] * num_images_per_prompt
									if isinstance(UpperCamelCase__		,					UpperCamelCase__						):
											lowerCAmelCase					:					List[Any]								=							torch.cat(UpperCamelCase__		,					dim=0						)

									if do_classifier_free_guidance:
											lowerCAmelCase					:					List[Any]								=							image_embeds.repeat_interleave(UpperCamelCase__		,					dim=0						)
											lowerCAmelCase					:					str								=							negative_image_embeds.repeat_interleave(UpperCamelCase__		,					dim=0						)

									lowerCAmelCase					:					Optional[int]								=							torch.cat([negative_image_embeds, image_embeds]		,					dim=0						).to(dtype=self.unet.dtype		,					device=UpperCamelCase__						)

									self.scheduler.set_timesteps(UpperCamelCase__		,					device=UpperCamelCase__						)
									lowerCAmelCase					:					Tuple								=							self.scheduler.timesteps

									lowerCAmelCase					:					str								=							self.unet.config.in_channels

									lowerCAmelCase						,  lowerCAmelCase					:					Tuple								=							downscale_height_and_width(UpperCamelCase__		,					UpperCamelCase__		,					self.movq_scale_factor						)

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

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

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

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

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

									if not return_dict:
											return (image,)

									return ImagePipelineOutput(images=UpperCamelCase__						)









"""simple docstring"""







import baseaa
def  a__      (    SCREAMING_SNAKE_CASE   :   str    ):


  '''simple docstring'''







  return baseaa.baaencode(string.encode("utf-8"    )    )
def  a__      (    SCREAMING_SNAKE_CASE   :   bytes    ):


  '''simple docstring'''







  return baseaa.baadecode(SCREAMING_SNAKE_CASE    ).decode("utf-8"    )


if __name__ == "__main__":
     lowerCAmelCase__				    =       '''Hello World!'''
     lowerCAmelCase__				    =       baseaa_encode(test)
     print(encoded)

     lowerCAmelCase__				    =       baseaa_decode(encoded)
     print(decoded)





