
from typing import List, Optional, Union

import numpy as np
import PIL
import torch
from PIL import Image

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    :      Union[str, Any]          =			logging.get_logger(__name__)  # pylint: disable=invalid-name

_SCREAMING_SNAKE_CASE    :      Any          =			'\n    Examples:\n        ```py\n        >>> from diffusers import KandinskyV22Img2ImgPipeline, KandinskyV22PriorPipeline\n        >>> from diffusers.utils import load_image\n        >>> import torch\n\n        >>> pipe_prior = KandinskyV22PriorPipeline.from_pretrained(\n        ...     "kandinsky-community/kandinsky-2-2-prior", torch_dtype=torch.float16\n        ... )\n        >>> pipe_prior.to("cuda")\n\n        >>> prompt = "A red cartoon frog, 4k"\n        >>> image_emb, zero_image_emb = pipe_prior(prompt, return_dict=False)\n\n        >>> pipe = KandinskyV22Img2ImgPipeline.from_pretrained(\n        ...     "kandinsky-community/kandinsky-2-2-decoder", torch_dtype=torch.float16\n        ... )\n        >>> pipe.to("cuda")\n\n        >>> init_image = load_image(\n        ...     "https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main"\n        ...     "/kandinsky/frog.png"\n        ... )\n\n        >>> image = pipe(\n        ...     image=init_image,\n        ...     image_embeds=image_emb,\n        ...     negative_image_embeds=zero_image_emb,\n        ...     height=768,\n        ...     width=768,\n        ...     num_inference_steps=100,\n        ...     strength=0.2,\n        ... ).images\n\n        >>> image[0].save("red_frog.png")\n        ```\n'

def 				SCREAMING_SNAKE_CASE     (      __UpperCamelCase :						List[Any]							,						__UpperCamelCase :						Tuple							,						__UpperCamelCase :						Optional[int]=8       ) ->       List[str]:

				"""simple docstring"""







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

def 				SCREAMING_SNAKE_CASE     (      __UpperCamelCase :						Any							,						__UpperCamelCase :						int=5_12							,						__UpperCamelCase :						Optional[Any]=5_12       ) ->       List[str]:

				"""simple docstring"""







				A__  :	str         =							pil_image.resize((w, h)							,						resample=Image.BICUBIC							,						reducing_gap=1       )
				A__  :	str         =							np.array(pil_image.convert('''RGB'''       )       )
				A__  :	str         =							arr.astype(np.floataa       ) / 1_27.5 - 1
				A__  :	List[str]         =							np.transpose(__UpperCamelCase							,						[2, 0, 1]       )
				A__  :	List[str]         =							torch.from_numpy(__UpperCamelCase       ).unsqueeze(0       )
				return image







class  UpperCamelCase__		(    SCREAMING_SNAKE_CASE_   ):







							'''simple docstring'''





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

											self.register_modules(
											    unet=UpperCamelCase__				,       scheduler=UpperCamelCase__				,       movq=UpperCamelCase__				,       )
											A__  :	Dict         =							2 ** (len(self.movq.config.block_out_channels		) - 1)





							def   __snake_case				(			self				,       UpperCamelCase__				,       UpperCamelCase__				,       UpperCamelCase__		):
											# get the original timestep using init_timestep
											A__  :	Tuple         =							min(int(num_inference_steps * strength		)				,       UpperCamelCase__		)

											A__  :	Optional[Any]         =							max(num_inference_steps - init_timestep				,       0		)
											A__  :	int         =							self.scheduler.timesteps[t_start:]

											return timesteps, num_inference_steps - t_start





							def   __snake_case				(			self				,       UpperCamelCase__				,       UpperCamelCase__				,       UpperCamelCase__				,       UpperCamelCase__				,       UpperCamelCase__				,       UpperCamelCase__				,       UpperCamelCase__=None		):
											if not isinstance(UpperCamelCase__				,       (torch.Tensor, PIL.Image.Image, list)		):
															raise ValueError(
															    F"`image` has to be of type `torch.Tensor`, `PIL.Image.Image` or list but is {type(UpperCamelCase__		)}"		)

											A__  :	List[str]         =							image.to(device=UpperCamelCase__				,       dtype=UpperCamelCase__		)

											A__  :	Dict         =							batch_size * num_images_per_prompt

											if image.shape[1] == 4:
															A__  :	int         =							image

											else:
															if isinstance(UpperCamelCase__				,       UpperCamelCase__		) and len(UpperCamelCase__		) != batch_size:
																			raise ValueError(
																			    F"You have passed a list of generators of length {len(UpperCamelCase__		)}, but requested an effective batch"
																			    F" size of {batch_size}. Make sure the batch size matches the length of the generators."		)

															elif isinstance(UpperCamelCase__				,       UpperCamelCase__		):
																			A__  :	Dict         =							[
																			    self.movq.encode(image[i : i + 1]		).latent_dist.sample(generator[i]		) for i in range(UpperCamelCase__		)
																			]
																			A__  :	int         =							torch.cat(UpperCamelCase__				,       dim=0		)
															else:
																			A__  :	str         =							self.movq.encode(UpperCamelCase__		).latent_dist.sample(UpperCamelCase__		)

															A__  :	int         =							self.movq.config.scaling_factor * init_latents

											A__  :	List[Any]         =							torch.cat([init_latents]				,       dim=0		)

											A__  :	str         =							init_latents.shape
											A__  :	str         =							randn_tensor(UpperCamelCase__				,       generator=UpperCamelCase__				,       device=UpperCamelCase__				,       dtype=UpperCamelCase__		)

											# get latents
											A__  :	str         =							self.scheduler.add_noise(UpperCamelCase__				,       UpperCamelCase__				,       UpperCamelCase__		)

											A__  :	str         =							init_latents

											return latents





							def   __snake_case				(			self				,       UpperCamelCase__=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__  :	List[Any]         =							[
											    self.unet,
											    self.movq,
											]
											for cpu_offloaded_model in models:
															if cpu_offloaded_model is not None:
																			cpu_offload(UpperCamelCase__				,       UpperCamelCase__		)





							def   __snake_case				(			self				,       UpperCamelCase__=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__  :	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)

											A__  :	Tuple         =							None
											for cpu_offloaded_model in [self.unet, self.movq]:
															A__  :	Tuple         =							cpu_offload_with_hook(UpperCamelCase__				,       UpperCamelCase__				,       prev_module_hook=UpperCamelCase__		)

											# We'll offload the last model manually.
											A__  :	int         =							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(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				,       UpperCamelCase__				,       UpperCamelCase__				,       UpperCamelCase__				,       UpperCamelCase__ = 512				,       UpperCamelCase__ = 512				,       UpperCamelCase__ = 100				,       UpperCamelCase__ = 4.0				,       UpperCamelCase__ = 0.3				,       UpperCamelCase__ = 1				,       UpperCamelCase__ = None				,       UpperCamelCase__ = "pil"				,       UpperCamelCase__ = True				,       ):
											A__  :	Any         =							self._execution_device

											A__  :	str         =							guidance_scale > 1.0

											if isinstance(UpperCamelCase__				,       UpperCamelCase__		):
															A__  :	Tuple         =							torch.cat(UpperCamelCase__				,       dim=0		)
											A__  :	Tuple         =							image_embeds.shape[0]
											if isinstance(UpperCamelCase__				,       UpperCamelCase__		):
															A__  :	List[Any]         =							torch.cat(UpperCamelCase__				,       dim=0		)

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

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

											if not isinstance(UpperCamelCase__				,       UpperCamelCase__		):
															A__  :	Optional[Any]         =							[image]
											if not all(isinstance(UpperCamelCase__				,       (PIL.Image.Image, torch.Tensor)		) for i in image		):
															raise ValueError(
															    F"Input is in incorrect format: {[type(UpperCamelCase__		) for i in image]}. Currently, we only support  PIL image and pytorch tensor"		)

											A__  :	str         =							torch.cat([prepare_image(UpperCamelCase__				,       UpperCamelCase__				,       UpperCamelCase__		) for i in image]				,       dim=0		)
											A__  :	List[Any]         =							image.to(dtype=image_embeds.dtype				,       device=UpperCamelCase__		)

											A__  :	List[str]         =							self.movq.encode(UpperCamelCase__		)['''latents''']
											A__  :	Dict         =							latents.repeat_interleave(UpperCamelCase__				,       dim=0		)
											self.scheduler.set_timesteps(UpperCamelCase__				,       device=UpperCamelCase__		)
											A__  :	Optional[int]         =							self.get_timesteps(UpperCamelCase__				,       UpperCamelCase__				,       UpperCamelCase__		)
											A__  :	List[Any]         =							timesteps[:1].repeat(batch_size * num_images_per_prompt		)
											A__  :	Any         =							downscale_height_and_width(UpperCamelCase__				,       UpperCamelCase__				,       self.movq_scale_factor		)
											A__  :	List[str]         =							self.prepare_latents(
											    UpperCamelCase__				,       UpperCamelCase__				,       UpperCamelCase__				,       UpperCamelCase__				,       image_embeds.dtype				,       UpperCamelCase__				,       UpperCamelCase__		)
											for i, t in enumerate(self.progress_bar(UpperCamelCase__		)		):
															# expand the latents if we are doing classifier free guidance
															A__  :	Tuple         =							torch.cat([latents] * 2		) if do_classifier_free_guidance else latents

															A__  :	List[str]         =							{'''image_embeds''': image_embeds}
															A__  :	Optional[int]         =							self.unet(
															    sample=UpperCamelCase__				,       timestep=UpperCamelCase__				,       encoder_hidden_states=UpperCamelCase__				,       added_cond_kwargs=UpperCamelCase__				,       return_dict=UpperCamelCase__				,       )[0]

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

															# compute the previous noisy sample x_t -> x_t-1
															A__  :	str         =							self.scheduler.step(
															    UpperCamelCase__				,       UpperCamelCase__				,       UpperCamelCase__				,       generator=UpperCamelCase__				,       )[0]

											# post-processing
											A__  :	List[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"]:
															A__  :	Any         =							image * 0.5 + 0.5
															A__  :	List[str]         =							image.clamp(0				,       1		)
															A__  :	Dict         =							image.cpu().permute(0				,       2				,       3				,       1		).float().numpy()

											if output_type == "pil":
															A__  :	Tuple         =							self.numpy_to_pil(UpperCamelCase__		)

											if not return_dict:
															return (image,)

											return ImagePipelineOutput(images=UpperCamelCase__		)

_SCREAMING_SNAKE_CASE    :      List[str]          =			range(2, 2_0 + 1)
_SCREAMING_SNAKE_CASE    :      Optional[Any]          =			[1_0**k for k in range(ks[-1] + 1)]
_SCREAMING_SNAKE_CASE    :      dict[int, dict[int, list[list[int]]]]          =			{}

def 				SCREAMING_SNAKE_CASE     (      __UpperCamelCase :						Any							,						__UpperCamelCase :						Dict							,						__UpperCamelCase :						Any							,						__UpperCamelCase :						Optional[Any]       ) ->       int:

				"""simple docstring"""







				A__  :	Tuple         =							sum(a_i[j] for j in range(__UpperCamelCase							,						len(__UpperCamelCase       )       )       )
				A__  :	Tuple         =							sum(a_i[j] * base[j] for j in range(min(len(__UpperCamelCase       )							,						__UpperCamelCase       )       )       )

				A__       ,							A__  :	Optional[int]         =							0, 0
				A__  :	List[Any]         =							n - i

				A__  :	Any         =							memo.get(__UpperCamelCase       )

				if sub_memo is not None:
								A__  :	Optional[int]         =							sub_memo.get(__UpperCamelCase       )

								if jumps is not None and len(__UpperCamelCase       ) > 0:
												# find and make the largest jump without going over
												A__  :	List[Any]         =							-1
												for _k in range(len(__UpperCamelCase       ) - 1							,						-1							,						-1       ):
																if jumps[_k][2] <= k and jumps[_k][1] <= max_dn:
																				A__  :	List[str]         =							_k
																				break

												if max_jump >= 0:
																A__       ,							A__       ,							A__  :	List[Any]         =							jumps[max_jump]
																# since the difference between jumps is cached, add c
																A__  :	int         =							diff + c
																for j in range(min(__UpperCamelCase							,						len(__UpperCamelCase       )       )       ):
																				A__       ,							A__  :	List[str]         =							divmod(__UpperCamelCase							,						10       )
																if new_c > 0:
																				add(__UpperCamelCase							,						__UpperCamelCase							,						__UpperCamelCase       )

								else:
												A__  :	List[Any]         =							[]
				else:
								A__  :	Optional[Any]         =							{c: []}
								A__  :	int         =							sub_memo

				if dn >= max_dn or c + diff >= base[k]:
								return diff, dn

				if k > ks[0]:
								while True:
												# keep doing smaller jumps
												A__       ,							A__  :	str         =							next_term(__UpperCamelCase							,						k - 1							,						i + dn							,						__UpperCamelCase       )
												diff += _diff
												dn += terms_jumped

												if dn >= max_dn or c + diff >= base[k]:
																break
				else:
								# would be too small a jump, just compute sequential terms instead
								A__       ,							A__  :	str         =							compute(__UpperCamelCase							,						__UpperCamelCase							,						i + dn							,						__UpperCamelCase       )
								diff += _diff
								dn += terms_jumped

				A__  :	str         =							sub_memo[c]

				# keep jumps sorted by # of terms skipped
				A__  :	List[Any]         =							0
				while j < len(__UpperCamelCase       ):
								if jumps[j][1] > dn:
												break
								j += 1

				# cache the jump for this value digitsum(b) and c
				sub_memo[c].insert(__UpperCamelCase							,						(diff, dn, k)       )
				return (diff, dn)

def 				SCREAMING_SNAKE_CASE     (      __UpperCamelCase :						int							,						__UpperCamelCase :						Optional[int]							,						__UpperCamelCase :						List[str]							,						__UpperCamelCase :						int       ) ->       Any:

				"""simple docstring"""







				if i >= n:
								return 0, i
				if k > len(__UpperCamelCase       ):
								a_i.extend([0 for _ in range(k - len(__UpperCamelCase       )       )]       )

				# note: a_i -> b * 10^k + c
				# ds_b -> digitsum(b)
				# ds_c -> digitsum(c)
				A__  :	Optional[Any]         =							i
				A__       ,							A__       ,							A__  :	Dict         =							0, 0, 0
				for j in range(len(__UpperCamelCase       )       ):
								if j >= k:
												ds_b += a_i[j]
								else:
												ds_c += a_i[j]

				while i < n:
								i += 1
								A__  :	int         =							ds_c + ds_b
								diff += addend
								A__  :	List[Any]         =							0
								for j in range(__UpperCamelCase       ):
												A__  :	Optional[Any]         =							a_i[j] + addend
												A__       ,							A__  :	List[str]         =							divmod(__UpperCamelCase							,						10       )

												ds_c += a_i[j]

								if addend > 0:
												break

				if addend > 0:
								add(__UpperCamelCase							,						__UpperCamelCase							,						__UpperCamelCase       )
				return diff, i - start_i

def 				SCREAMING_SNAKE_CASE     (      __UpperCamelCase :						Dict							,						__UpperCamelCase :						List[Any]							,						__UpperCamelCase :						int       ) ->       Tuple:

				"""simple docstring"""







				for j in range(__UpperCamelCase							,						len(__UpperCamelCase       )       ):
								A__  :	Any         =							digits[j] + addend
								if s >= 10:
												A__       ,							A__  :	Union[str, Any]         =							divmod(__UpperCamelCase							,						10       )
												A__  :	Optional[int]         =							addend // 10 + quotient
								else:
												A__  :	Any         =							s
												A__  :	Dict         =							addend // 10

								if addend == 0:
												break

				while addend > 0:
								A__       ,							A__  :	Dict         =							divmod(__UpperCamelCase							,						10       )
								digits.append(__UpperCamelCase       )

def 				SCREAMING_SNAKE_CASE     (      __UpperCamelCase :						int = 10**15       ) ->       int:

				"""simple docstring"""







				A__  :	List[Any]         =							[1]
				A__  :	Dict         =							1
				A__  :	Tuple         =							0
				while True:
								A__       ,							A__  :	List[str]         =							next_term(__UpperCamelCase							,						20							,						i + dn							,						__UpperCamelCase       )
								dn += terms_jumped
								if dn == n - i:
												break

				A__  :	List[str]         =							0
				for j in range(len(__UpperCamelCase       )       ):
								a_n += digits[j] * 10**j
				return a_n


if __name__ == "__main__":
					print(f"""{solution() = }""")