


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     :					Dict      =       logging.get_logger(__name__)  # pylint: disable=invalid-name

_SCREAMING_SNAKE_CASE     :					List[str]      =       "\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      UpperCAmelCase__		(UpperCamelCase_				,UpperCamelCase_				,UpperCamelCase_=8							):



     """simple docstring"""






     snake_case											=							height // scale_factor**2
     if height % scale_factor**2 != 0:
          new_height += 1
     snake_case											=							width // scale_factor**2
     if width % scale_factor**2 != 0:
          new_width += 1
     return new_height * scale_factor, new_width * scale_factor
def      UpperCAmelCase__		(UpperCamelCase_				,UpperCamelCase_=5_12				,UpperCamelCase_=5_12							):



     """simple docstring"""






     snake_case											=							pil_image.resize((w, h)				,resample=Image.BICUBIC				,reducing_gap=1							)
     snake_case											=							np.array(pil_image.convert('''RGB'''							)							)
     snake_case											=							arr.astype(np.floataa							) / 127.5 - 1
     snake_case											=							np.transpose(UpperCamelCase_				,[2, 0, 1]							)
     snake_case											=							torch.from_numpy(UpperCamelCase_							).unsqueeze(0							)
     return image

class     A__   (       snake_case__      ):







    """simple docstring"""

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

         self.register_modules(
             unet=__snake_case     ,   scheduler=__snake_case     ,   movq=__snake_case     ,   )
         snake_case											=							2 ** (len(self.movq.config.block_out_channels       ) - 1)
    def 						a_     (			self     ,   __snake_case     ,   __snake_case     ,   __snake_case       ):
         # get the original timestep using init_timestep
         snake_case											=							min(int(num_inference_steps * strength       )     ,   __snake_case       )

         snake_case											=							max(num_inference_steps - init_timestep     ,   0       )
         snake_case											=							self.scheduler.timesteps[t_start:]

         return timesteps, num_inference_steps - t_start
    def 						a_     (			self     ,   __snake_case     ,   __snake_case     ,   __snake_case     ,   __snake_case     ,   __snake_case     ,   __snake_case     ,   __snake_case=None       ):
         if not isinstance(__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(__snake_case       )}'''       )

         snake_case											=							image.to(device=__snake_case     ,   dtype=__snake_case       )

         snake_case											=							batch_size * num_images_per_prompt

         if image.shape[1] == 4:
              snake_case											=							image

         else:
              if isinstance(__snake_case     ,   __snake_case       ) and len(__snake_case       ) != batch_size:
                   raise ValueError(
                       F'''You have passed a list of generators of length {len(__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(__snake_case     ,   __snake_case       ):
                   snake_case											=							[
                       self.movq.encode(image[i : i + 1]       ).latent_dist.sample(generator[i]       ) for i in range(__snake_case       )
                   ]
                   snake_case											=							torch.cat(__snake_case     ,   dim=0       )
              else:
                   snake_case											=							self.movq.encode(__snake_case       ).latent_dist.sample(__snake_case       )

              snake_case											=							self.movq.config.scaling_factor * init_latents

         snake_case											=							torch.cat([init_latents]     ,   dim=0       )

         snake_case											=							init_latents.shape
         snake_case											=							randn_tensor(__snake_case     ,   generator=__snake_case     ,   device=__snake_case     ,   dtype=__snake_case       )

         # get latents
         snake_case											=							self.scheduler.add_noise(__snake_case     ,   __snake_case     ,   __snake_case       )

         snake_case											=							init_latents

         return latents
    def 						a_     (			self     ,   __snake_case=0       ):
         if is_accelerate_available():
              from accelerate import cpu_offload
         else:
              raise ImportError('''Please install accelerate via `pip install accelerate`'''       )

         snake_case											=							torch.device(F'''cuda:{gpu_id}'''       )

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

         snake_case											=							torch.device(F'''cuda:{gpu_id}'''       )

         if self.device.type != "cpu":
              self.to('''cpu'''     ,   silence_dtype_warnings=__snake_case       )
              torch.cuda.empty_cache()  # otherwise we don't see the memory savings (but they probably exist)

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

         # We'll offload the last model manually.
         snake_case											=							hook
    @property
    # Copied from diffusers.pipelines.stable_diffusion.pipeline_stable_diffusion.StableDiffusionPipeline._execution_device
    def 						a_     (			self       ):
         if not hasattr(self.unet     ,   '''_hf_hook'''       ):
              return self.device
         for module in self.unet.modules():
              if (
                  hasattr(__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(__snake_case       )
    def __call__(			self     ,   __snake_case     ,   __snake_case     ,   __snake_case     ,   __snake_case = 5_1_2     ,   __snake_case = 5_1_2     ,   __snake_case = 1_0_0     ,   __snake_case = 4.0     ,   __snake_case = 0.3     ,   __snake_case = 1     ,   __snake_case = None     ,   __snake_case = "pil"     ,   __snake_case = True     ,   ):
         snake_case											=							self._execution_device

         snake_case											=							guidance_scale > 1.0

         if isinstance(__snake_case     ,   __snake_case       ):
              snake_case											=							torch.cat(__snake_case     ,   dim=0       )
         snake_case											=							image_embeds.shape[0]
         if isinstance(__snake_case     ,   __snake_case       ):
              snake_case											=							torch.cat(__snake_case     ,   dim=0       )

         if do_classifier_free_guidance:
              snake_case											=							image_embeds.repeat_interleave(__snake_case     ,   dim=0       )
              snake_case											=							negative_image_embeds.repeat_interleave(__snake_case     ,   dim=0       )

         snake_case											=							torch.cat([negative_image_embeds, image_embeds]     ,   dim=0       ).to(dtype=self.unet.dtype     ,   device=__snake_case       )

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

         snake_case											=							torch.cat([prepare_image(__snake_case     ,   __snake_case     ,   __snake_case       ) for i in image]     ,   dim=0       )
         snake_case											=							image.to(dtype=image_embeds.dtype     ,   device=__snake_case       )

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

              snake_case											=							{'''image_embeds''': image_embeds}
              snake_case											=							self.unet(
                  sample=__snake_case     ,   timestep=__snake_case     ,   encoder_hidden_states=__snake_case     ,   added_cond_kwargs=__snake_case     ,   return_dict=__snake_case     ,   )[0]

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

              # compute the previous noisy sample x_t -> x_t-1
              snake_case											=							self.scheduler.step(
                  __snake_case     ,   __snake_case     ,   __snake_case     ,   generator=__snake_case     ,   )[0]

         # post-processing
         snake_case											=							self.movq.decode(__snake_case     ,   force_not_quantize=__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"]:
              snake_case											=							image * 0.5 + 0.5
              snake_case											=							image.clamp(0     ,   1       )
              snake_case											=							image.cpu().permute(0     ,   2     ,   3     ,   1       ).float().numpy()

         if output_type == "pil":
              snake_case											=							self.numpy_to_pil(__snake_case       )

         if not return_dict:
              return (image,)

         return ImagePipelineOutput(images=__snake_case       )








from functools import lru_cache
def      UpperCAmelCase__		(UpperCamelCase_							):



     """simple docstring"""






     snake_case											=							2
     snake_case											=							set()
     while i * i <= n:
          if n % i:
               i += 1
          else:
               n //= i
               factors.add(UpperCamelCase_							)
     if n > 1:
          factors.add(UpperCamelCase_							)
     return factors
@lru_cache
def      UpperCAmelCase__		(UpperCamelCase_							):



     """simple docstring"""






     return len(unique_prime_factors(UpperCamelCase_							)							)
def      UpperCAmelCase__		(UpperCamelCase_							):



     """simple docstring"""






     return len(set(UpperCamelCase_							)							) in (0, 1)
def      UpperCAmelCase__		(UpperCamelCase_							):



     """simple docstring"""






     snake_case											=							2

     while True:
          # Increment each value of a generated range
          snake_case											=							[base + i for i in range(UpperCamelCase_							)]

          # Run elements through out unique_prime_factors function
          # Append our target number to the end.
          snake_case											=							[upf_len(UpperCamelCase_							) for x in group]
          checker.append(UpperCamelCase_							)

          # If all numbers in the list are equal, return the group variable.
          if equality(UpperCamelCase_							):
               return group

          # Increment our base variable by 1
          base += 1
def      UpperCAmelCase__		(UpperCamelCase_ = 4							):



     """simple docstring"""






     snake_case											=							run(UpperCamelCase_							)
     return results[0] if len(UpperCamelCase_							) else None


if __name__ == "__main__":
 print(solution())




