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


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

UpperCAmelCase        =				"""
    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 __magic_name__	(				_lowerCamelCase:		str,					_lowerCamelCase:		Dict,					_lowerCamelCase:		Dict=8			)					->						Any:

	'''simple docstring'''




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



class 			lowercase					( lowercase__		):






			def __init__(self  :	Optional[Any]      ,SCREAMING_SNAKE_CASE_  :	UNetaDConditionModel      ,SCREAMING_SNAKE_CASE_  :	DDPMScheduler      ,SCREAMING_SNAKE_CASE_  :	VQModel      ,)       ->	Any:





				"""simple docstring"""


				super().__init__()

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






			def 					UpperCAmelCase							(self  :	Optional[int]      ,SCREAMING_SNAKE_CASE_  :	str      ,SCREAMING_SNAKE_CASE_  :	List[Any]      ,SCREAMING_SNAKE_CASE_  :	Any      ,SCREAMING_SNAKE_CASE_  :	Any      ,SCREAMING_SNAKE_CASE_  :	Tuple      ,SCREAMING_SNAKE_CASE_  :	Union[str, Any]			)       ->	Optional[int]:





				"""simple docstring"""


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

				lowerCAmelCase												=  latents * scheduler.init_noise_sigma
				return latents






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





				"""simple docstring"""


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

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

				lowerCAmelCase												=  [
				    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 					UpperCAmelCase							(self  :	int      ,SCREAMING_SNAKE_CASE_  :	Optional[int]=0			)       ->	int:





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

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

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






			@property
			# Copied from diffusers.pipelines.stable_diffusion.pipeline_stable_diffusion.StableDiffusionPipeline._execution_device
			def 					UpperCAmelCase							(self  :	int			)       ->	List[Any]:





				"""simple docstring"""


				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  :	Optional[int]      ,SCREAMING_SNAKE_CASE_  :	Union[torch.FloatTensor, List[torch.FloatTensor]]      ,SCREAMING_SNAKE_CASE_  :	Union[torch.FloatTensor, List[torch.FloatTensor]]      ,SCREAMING_SNAKE_CASE_  :	int = 512      ,SCREAMING_SNAKE_CASE_  :	int = 512      ,SCREAMING_SNAKE_CASE_  :	int = 100      ,SCREAMING_SNAKE_CASE_  :	float = 4.0      ,SCREAMING_SNAKE_CASE_  :	int = 1      ,SCREAMING_SNAKE_CASE_  :	Optional[Union[torch.Generator, List[torch.Generator]]] = None      ,SCREAMING_SNAKE_CASE_  :	Optional[torch.FloatTensor] = None      ,SCREAMING_SNAKE_CASE_  :	Optional[str] = "pil"      ,SCREAMING_SNAKE_CASE_  :	bool = True      ,)       ->	Union[str, Any]:





				"""simple docstring"""


				lowerCAmelCase												=  self._execution_device

				lowerCAmelCase												=  guidance_scale > 1.0

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

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

				lowerCAmelCase												=  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_			)
				lowerCAmelCase												=  self.scheduler.timesteps

				lowerCAmelCase												=  self.unet.config.in_channels

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

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

					lowerCAmelCase												=  {'''image_embeds''': image_embeds}
					lowerCAmelCase												=  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:
						lowerCAmelCase							,       lowerCAmelCase												=  noise_pred.split(latents.shape[1]      ,dim=1			)
						lowerCAmelCase							,       lowerCAmelCase												=  noise_pred.chunk(2			)
						lowerCAmelCase							,       lowerCAmelCase												=  variance_pred.chunk(2			)
						lowerCAmelCase												=  noise_pred_uncond + guidance_scale * (noise_pred_text - noise_pred_uncond)
						lowerCAmelCase												=  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												=  noise_pred.split(latents.shape[1]      ,dim=1			)

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

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

				if not return_dict:
					return (image,)

				return ImagePipelineOutput(images=SCREAMING_SNAKE_CASE_			)




"""simple docstring"""






from __future__ import annotations



class 			lowercase					:






			def __init__(self  :	List[Any]      ,SCREAMING_SNAKE_CASE_  :	int = 0			)       ->	List[Any]:





				"""simple docstring"""


				lowerCAmelCase												=  key






			def 					UpperCAmelCase							(self  :	Tuple      ,SCREAMING_SNAKE_CASE_  :	str      ,SCREAMING_SNAKE_CASE_  :	int			)       ->	list[str]:





				"""simple docstring"""


				assert isinstance(SCREAMING_SNAKE_CASE_      ,SCREAMING_SNAKE_CASE_			) and isinstance(SCREAMING_SNAKE_CASE_      ,SCREAMING_SNAKE_CASE_			)

				lowerCAmelCase												=  key or self.__key or 1

				# make sure key is an appropriate size
				key %= 255

				return [chr(ord(SCREAMING_SNAKE_CASE_			) ^ key			) for ch in content]






			def 					UpperCAmelCase							(self  :	int      ,SCREAMING_SNAKE_CASE_  :	str      ,SCREAMING_SNAKE_CASE_  :	int			)       ->	list[str]:





				"""simple docstring"""


				assert isinstance(SCREAMING_SNAKE_CASE_      ,SCREAMING_SNAKE_CASE_			) and isinstance(SCREAMING_SNAKE_CASE_      ,SCREAMING_SNAKE_CASE_			)

				lowerCAmelCase												=  key or self.__key or 1

				# make sure key is an appropriate size
				key %= 255

				return [chr(ord(SCREAMING_SNAKE_CASE_			) ^ key			) for ch in content]






			def 					UpperCAmelCase							(self  :	Union[str, Any]      ,SCREAMING_SNAKE_CASE_  :	str      ,SCREAMING_SNAKE_CASE_  :	int = 0			)       ->	str:





				"""simple docstring"""


				assert isinstance(SCREAMING_SNAKE_CASE_      ,SCREAMING_SNAKE_CASE_			) and isinstance(SCREAMING_SNAKE_CASE_      ,SCREAMING_SNAKE_CASE_			)

				lowerCAmelCase												=  key or self.__key or 1

				# make sure key can be any size
				while key > 255:
					key -= 255

				# This will be returned
				lowerCAmelCase												=  ''''''

				for ch in content:
					ans += chr(ord(SCREAMING_SNAKE_CASE_			) ^ key			)

				return ans






			def 					UpperCAmelCase							(self  :	Union[str, Any]      ,SCREAMING_SNAKE_CASE_  :	str      ,SCREAMING_SNAKE_CASE_  :	int = 0			)       ->	str:





				"""simple docstring"""


				assert isinstance(SCREAMING_SNAKE_CASE_      ,SCREAMING_SNAKE_CASE_			) and isinstance(SCREAMING_SNAKE_CASE_      ,SCREAMING_SNAKE_CASE_			)

				lowerCAmelCase												=  key or self.__key or 1

				# make sure key can be any size
				while key > 255:
					key -= 255

				# This will be returned
				lowerCAmelCase												=  ''''''

				for ch in content:
					ans += chr(ord(SCREAMING_SNAKE_CASE_			) ^ key			)

				return ans






			def 					UpperCAmelCase							(self  :	List[str]      ,SCREAMING_SNAKE_CASE_  :	str      ,SCREAMING_SNAKE_CASE_  :	int = 0			)       ->	bool:





				"""simple docstring"""


				assert isinstance(SCREAMING_SNAKE_CASE_      ,SCREAMING_SNAKE_CASE_			) and isinstance(SCREAMING_SNAKE_CASE_      ,SCREAMING_SNAKE_CASE_			)

				try:
					with open(SCREAMING_SNAKE_CASE_			) as fin, open('''encrypt.out'''      ,'''w+'''			) as fout:
						# actual encrypt-process
						for line in fin:
							fout.write(self.encrypt_string(SCREAMING_SNAKE_CASE_      ,SCREAMING_SNAKE_CASE_			)			)

				except OSError:
					return False

				return True






			def 					UpperCAmelCase							(self  :	List[Any]      ,SCREAMING_SNAKE_CASE_  :	str      ,SCREAMING_SNAKE_CASE_  :	int			)       ->	bool:





				"""simple docstring"""


				assert isinstance(SCREAMING_SNAKE_CASE_      ,SCREAMING_SNAKE_CASE_			) and isinstance(SCREAMING_SNAKE_CASE_      ,SCREAMING_SNAKE_CASE_			)

				try:
					with open(SCREAMING_SNAKE_CASE_			) as fin, open('''decrypt.out'''      ,'''w+'''			) as fout:
						# actual encrypt-process
						for line in fin:
							fout.write(self.decrypt_string(SCREAMING_SNAKE_CASE_      ,SCREAMING_SNAKE_CASE_			)			)

				except OSError:
					return False

				return True


# Tests
# crypt = XORCipher()
# key = 67

# # test encrypt
# print(crypt.encrypt("hallo welt",key))
# # test decrypt
# print(crypt.decrypt(crypt.encrypt("hallo welt",key), key))

# # test encrypt_string
# print(crypt.encrypt_string("hallo welt",key))

# # test decrypt_string
# print(crypt.decrypt_string(crypt.encrypt_string("hallo welt",key),key))

# if (crypt.encrypt_file("test.txt",key)):
#       print("encrypt successful")
# else:
#       print("encrypt unsuccessful")

# if (crypt.decrypt_file("encrypt.out",key)):
#       print("decrypt successful")
# else:
#       print("decrypt unsuccessful")


