



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


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

a	:   Optional[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       lowerCAmelCase_     (lowerCAmelCase__:						int						,		lowerCAmelCase__:						int						,		lowerCAmelCase__:						Dict=8   ):





			"""simple docstring"""


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







def       lowerCAmelCase_     (lowerCAmelCase__:						Union[str, Any]						,		lowerCAmelCase__:						str=5_1_2						,		lowerCAmelCase__:						str=5_1_2   ):





			"""simple docstring"""


			UpperCAmelCase_:						Union[str, Any]					=			pil_image.resize((w, h)						,		resample=Image.BICUBIC						,		reducing_gap=1   )
			UpperCAmelCase_:						int					=			np.array(pil_image.convert("""RGB"""   )   )
			UpperCAmelCase_:						Optional[Any]					=			arr.astype(np.floataa   ) / 127.5 - 1
			UpperCAmelCase_:						List[str]					=			np.transpose(lowerCAmelCase__						,		[2, 0, 1]   )
			UpperCAmelCase_:						Optional[Any]					=			torch.from_numpy(lowerCAmelCase__   ).unsqueeze(0   )
			return image







class 							_a							(					UpperCAmelCase__ ):

			def __init__(self,					SCREAMING_SNAKE_CASE_,					SCREAMING_SNAKE_CASE_,					SCREAMING_SNAKE_CASE_,					)    ->					Dict:
						super().__init__()

						self.register_modules(
						    unet=_SCREAMING_SNAKE_CASE,					scheduler=_SCREAMING_SNAKE_CASE,					movq=_SCREAMING_SNAKE_CASE,					)
						UpperCAmelCase_:						Any					=			2 ** (len(self.movq.config.block_out_channels ) - 1)

			def 			__snake_case    (self,					SCREAMING_SNAKE_CASE_,					SCREAMING_SNAKE_CASE_,					SCREAMING_SNAKE_CASE_ )    ->					List[Any]:
						UpperCAmelCase_:						Union[str, Any]					=			min(int(num_inference_steps * strength ),					_SCREAMING_SNAKE_CASE )

						UpperCAmelCase_:						str					=			max(num_inference_steps - init_timestep,					0 )
						UpperCAmelCase_:						Union[str, Any]					=			self.scheduler.timesteps[t_start:]

						return timesteps, num_inference_steps - t_start

			def 			__snake_case    (self,					SCREAMING_SNAKE_CASE_,					SCREAMING_SNAKE_CASE_,					SCREAMING_SNAKE_CASE_,					SCREAMING_SNAKE_CASE_,					SCREAMING_SNAKE_CASE_,					SCREAMING_SNAKE_CASE_,					SCREAMING_SNAKE_CASE_=None )    ->					str:




						if not isinstance(_SCREAMING_SNAKE_CASE,					(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(_SCREAMING_SNAKE_CASE )}' )

						UpperCAmelCase_:						List[str]					=			image.to(device=_SCREAMING_SNAKE_CASE,					dtype=_SCREAMING_SNAKE_CASE )

						UpperCAmelCase_:						Optional[Any]					=			batch_size * num_images_per_prompt

						if image.shape[1] == 4:
									UpperCAmelCase_:						str					=			image

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

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

									UpperCAmelCase_:						Tuple					=			self.movq.config.scaling_factor * init_latents

						UpperCAmelCase_:						Tuple					=			torch.cat([init_latents],					dim=0 )

						UpperCAmelCase_:						int					=			init_latents.shape
						UpperCAmelCase_:						Union[str, Any]					=			randn_tensor(_SCREAMING_SNAKE_CASE,					generator=_SCREAMING_SNAKE_CASE,					device=_SCREAMING_SNAKE_CASE,					dtype=_SCREAMING_SNAKE_CASE )

						# get latents
						UpperCAmelCase_:						Optional[Any]					=			self.scheduler.add_noise(_SCREAMING_SNAKE_CASE,					_SCREAMING_SNAKE_CASE,					_SCREAMING_SNAKE_CASE )

						UpperCAmelCase_:						List[Any]					=			init_latents

						return latents

			def 			__snake_case    (self,					SCREAMING_SNAKE_CASE_=0 )    ->					Optional[int]:




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

						UpperCAmelCase_:						Any					=			torch.device(f'cuda:{gpu_id}' )

						UpperCAmelCase_:						Union[str, Any]					=			[
						    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,					SCREAMING_SNAKE_CASE_=0 )    ->					Tuple:




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

						UpperCAmelCase_:						str					=			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)

						UpperCAmelCase_:						Dict					=			None
						for cpu_offloaded_model in [self.unet, self.movq]:
									UpperCAmelCase_:						int					=			cpu_offload_with_hook(_SCREAMING_SNAKE_CASE,					_SCREAMING_SNAKE_CASE,					prev_module_hook=_SCREAMING_SNAKE_CASE )

						# We'll offload the last model manually.
						UpperCAmelCase_:						List[Any]					=			hook

			@property
			# Copied from diffusers.pipelines.stable_diffusion.pipeline_stable_diffusion.StableDiffusionPipeline._execution_device
			def 			__snake_case    (self )    ->					Dict:




						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,					SCREAMING_SNAKE_CASE_,					SCREAMING_SNAKE_CASE_,					SCREAMING_SNAKE_CASE_,					SCREAMING_SNAKE_CASE_ = 512,					SCREAMING_SNAKE_CASE_ = 512,					SCREAMING_SNAKE_CASE_ = 100,					SCREAMING_SNAKE_CASE_ = 4.0,					SCREAMING_SNAKE_CASE_ = 0.3,					SCREAMING_SNAKE_CASE_ = 1,					SCREAMING_SNAKE_CASE_ = None,					SCREAMING_SNAKE_CASE_ = "pil",					SCREAMING_SNAKE_CASE_ = True,					)    ->					Tuple:
						UpperCAmelCase_:						Any					=			self._execution_device

						UpperCAmelCase_:						Union[str, Any]					=			guidance_scale > 1.0

						if isinstance(_SCREAMING_SNAKE_CASE,					_SCREAMING_SNAKE_CASE ):
									UpperCAmelCase_:						int					=			torch.cat(_SCREAMING_SNAKE_CASE,					dim=0 )
						UpperCAmelCase_:						Any					=			image_embeds.shape[0]
						if isinstance(_SCREAMING_SNAKE_CASE,					_SCREAMING_SNAKE_CASE ):
									UpperCAmelCase_:						str					=			torch.cat(_SCREAMING_SNAKE_CASE,					dim=0 )

						if do_classifier_free_guidance:
									UpperCAmelCase_:						int					=			image_embeds.repeat_interleave(_SCREAMING_SNAKE_CASE,					dim=0 )
									UpperCAmelCase_:						List[Any]					=			negative_image_embeds.repeat_interleave(_SCREAMING_SNAKE_CASE,					dim=0 )

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

						if not isinstance(_SCREAMING_SNAKE_CASE,					_SCREAMING_SNAKE_CASE ):
									UpperCAmelCase_:						Tuple					=			[image]
						if not all(isinstance(_SCREAMING_SNAKE_CASE,					(PIL.Image.Image, torch.Tensor) ) for i in image ):
									raise ValueError(
									    f'Input is in incorrect format: {[type(_SCREAMING_SNAKE_CASE ) for i in image]}. Currently, we only support  PIL image and pytorch tensor' )

						UpperCAmelCase_:						Optional[Any]					=			torch.cat([prepare_image(_SCREAMING_SNAKE_CASE,					_SCREAMING_SNAKE_CASE,					_SCREAMING_SNAKE_CASE ) for i in image],					dim=0 )
						UpperCAmelCase_:						Any					=			image.to(dtype=image_embeds.dtype,					device=_SCREAMING_SNAKE_CASE )

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

									UpperCAmelCase_:						Union[str, Any]					=			{"""image_embeds""": image_embeds}
									UpperCAmelCase_:						Union[str, Any]					=			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:
												UpperCAmelCase_:						Dict					=			noise_pred.split(latents.shape[1],					dim=1 )
												UpperCAmelCase_:						Union[str, Any]					=			noise_pred.chunk(2 )
												UpperCAmelCase_:						Any					=			variance_pred.chunk(2 )
												UpperCAmelCase_:						Tuple					=			noise_pred_uncond + guidance_scale * (noise_pred_text - noise_pred_uncond)
												UpperCAmelCase_:						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"]
									):
												UpperCAmelCase_:						int					=			noise_pred.split(latents.shape[1],					dim=1 )

									# compute the previous noisy sample x_t -> x_t-1
									UpperCAmelCase_:						List[Any]					=			self.scheduler.step(
									    _SCREAMING_SNAKE_CASE,					_SCREAMING_SNAKE_CASE,					_SCREAMING_SNAKE_CASE,					generator=_SCREAMING_SNAKE_CASE,					)[0]

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

						if output_type == "pil":
									UpperCAmelCase_:						Tuple					=			self.numpy_to_pil(_SCREAMING_SNAKE_CASE )

						if not return_dict:
									return (image,)

						return ImagePipelineOutput(images=_SCREAMING_SNAKE_CASE )











"""simple docstring"""
from math import isqrt, loga

def       _snake_case						(						UpperCamelCase						:		int						):
 UpperCAmelCase	:					Any	=	[True] * max_number
 for i in range(2			,		isqrt(max_number - 1						) + 1						):
  if is_prime[i]:
   for j in range(i**2			,		UpperCamelCase			,		UpperCamelCase						):
    UpperCAmelCase	:					str	=	False

 return [i for i in range(2			,		UpperCamelCase						) if is_prime[i]]

def       _snake_case						(						UpperCamelCase						:		int = 800800			,		UpperCamelCase						:		int = 800800						):
 UpperCAmelCase	:					Union[str, Any]	=	degree * loga(UpperCamelCase						)
 UpperCAmelCase	:					int	=	int(UpperCamelCase						)
 UpperCAmelCase	:					Union[str, Any]	=	calculate_prime_numbers(UpperCamelCase						)

 UpperCAmelCase	:					Dict	=	0
 UpperCAmelCase	:					Optional[int]	=	0
 UpperCAmelCase	:					Dict	=	len(UpperCamelCase						) - 1
 while left < right:
  while (
      prime_numbers[right] * loga(prime_numbers[left]						)
      + prime_numbers[left] * loga(prime_numbers[right]						)
      > upper_bound
  ):
   right -= 1
  hybrid_integers_count += right - left
  left += 1

 return hybrid_integers_count


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