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


lowercase_  : List[Any]								=					logging.get_logger(__name__)  # pylint: disable=invalid-name

lowercase_  : 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       A__   (						snake_case_			:		List[Any]							,					snake_case_			:		str							,					snake_case_			:		str=8   ):
 SCREAMING_SNAKE_CASE__: str=   height // scale_factor**2
 if height % scale_factor**2 != 0:
  new_height += 1
 SCREAMING_SNAKE_CASE__: Tuple=   width // scale_factor**2
 if width % scale_factor**2 != 0:
  new_width += 1
 return new_height * scale_factor, new_width * scale_factor







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






class 		_lowerCamelCase		( a__     ):
    def __init__(	self			,			lowerCAmelCase			,			lowerCAmelCase			,			lowerCAmelCase			,			)							->       str:
     super().__init__()

     self.register_modules(
         unet=__a			,			scheduler=__a			,			movq=__a			,			)
     SCREAMING_SNAKE_CASE__: List[str]=   2 ** (len(self.movq.config.block_out_channels ) - 1)
    def     UpperCamelCase_					(	self			,			lowerCAmelCase			,			lowerCAmelCase			,			lowerCAmelCase )							->       Tuple:
     # get the original timestep using init_timestep
     SCREAMING_SNAKE_CASE__: List[Any]=   min(int(num_inference_steps * strength )			,			__a )

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

     return timesteps, num_inference_steps - t_start
    def     UpperCamelCase_					(	self			,			lowerCAmelCase			,			lowerCAmelCase			,			lowerCAmelCase			,			lowerCAmelCase			,			lowerCAmelCase			,			lowerCAmelCase			,			lowerCAmelCase=None )							->       Tuple:
     if not isinstance(__a			,			(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(__a )}' )

     SCREAMING_SNAKE_CASE__: List[str]=   image.to(device=__a			,			dtype=__a )

     SCREAMING_SNAKE_CASE__: Optional[Any]=   batch_size * num_images_per_prompt

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

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

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

      SCREAMING_SNAKE_CASE__: List[Any]=   self.movq.config.scaling_factor * init_latents

     SCREAMING_SNAKE_CASE__: str=   torch.cat([init_latents]			,			dim=0 )

     SCREAMING_SNAKE_CASE__: Optional[Any]=   init_latents.shape
     SCREAMING_SNAKE_CASE__: Dict=   randn_tensor(__a			,			generator=__a			,			device=__a			,			dtype=__a )

     # get latents
     SCREAMING_SNAKE_CASE__: Dict=   self.scheduler.add_noise(__a			,			__a			,			__a )

     SCREAMING_SNAKE_CASE__: List[Any]=   init_latents

     return latents
    def     UpperCamelCase_					(	self			,			lowerCAmelCase=0 )							->       Optional[Any]:
     if is_accelerate_available():
      from accelerate import cpu_offload
     else:
      raise ImportError('''Please install accelerate via `pip install accelerate`''' )

     SCREAMING_SNAKE_CASE__: Union[str, Any]=   torch.device(f'cuda:{gpu_id}' )

     SCREAMING_SNAKE_CASE__: int=   [
         self.unet,
         self.movq,
     ]
     for cpu_offloaded_model in models:
      if cpu_offloaded_model is not None:
       cpu_offload(__a			,			__a )
    def     UpperCamelCase_					(	self			,			lowerCAmelCase=0 )							->       Union[str, Any]:
     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.''' )

     SCREAMING_SNAKE_CASE__: str=   torch.device(f'cuda:{gpu_id}' )

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

     SCREAMING_SNAKE_CASE__: List[Any]=   None
     for cpu_offloaded_model in [self.unet, self.movq]:
      SCREAMING_SNAKE_CASE__: Any=   cpu_offload_with_hook(__a			,			__a			,			prev_module_hook=__a )

     # We'll offload the last model manually.
     SCREAMING_SNAKE_CASE__: List[Any]=   hook
    @property
    # Copied from diffusers.pipelines.stable_diffusion.pipeline_stable_diffusion.StableDiffusionPipeline._execution_device
    def     UpperCamelCase_					(	self )							->       Optional[Any]:
     if not hasattr(self.unet			,			'''_hf_hook''' ):
      return self.device
     for module in self.unet.modules():
      if (
          hasattr(__a			,			'''_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(__a )
    def __call__(	self			,			lowerCAmelCase			,			lowerCAmelCase			,			lowerCAmelCase			,			lowerCAmelCase = 512			,			lowerCAmelCase = 512			,			lowerCAmelCase = 100			,			lowerCAmelCase = 4.0			,			lowerCAmelCase = 0.3			,			lowerCAmelCase = 1			,			lowerCAmelCase = None			,			lowerCAmelCase = "pil"			,			lowerCAmelCase = True			,			)							->       int:
     SCREAMING_SNAKE_CASE__: Optional[Any]=   self._execution_device

     SCREAMING_SNAKE_CASE__: Any=   guidance_scale > 1.0

     if isinstance(__a			,			__a ):
      SCREAMING_SNAKE_CASE__: Union[str, Any]=   torch.cat(__a			,			dim=0 )
     SCREAMING_SNAKE_CASE__: Optional[Any]=   image_embeds.shape[0]
     if isinstance(__a			,			__a ):
      SCREAMING_SNAKE_CASE__: List[Any]=   torch.cat(__a			,			dim=0 )

     if do_classifier_free_guidance:
      SCREAMING_SNAKE_CASE__: str=   image_embeds.repeat_interleave(__a			,			dim=0 )
      SCREAMING_SNAKE_CASE__: str=   negative_image_embeds.repeat_interleave(__a			,			dim=0 )

     SCREAMING_SNAKE_CASE__: str=   torch.cat([negative_image_embeds, image_embeds]			,			dim=0 ).to(dtype=self.unet.dtype			,			device=__a )

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

     SCREAMING_SNAKE_CASE__: int=   torch.cat([prepare_image(__a			,			__a			,			__a ) for i in image]			,			dim=0 )
     SCREAMING_SNAKE_CASE__: Dict=   image.to(dtype=image_embeds.dtype			,			device=__a )

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

      SCREAMING_SNAKE_CASE__: Tuple=   {"""image_embeds""": image_embeds}
      SCREAMING_SNAKE_CASE__: Dict=   self.unet(
          sample=__a			,			timestep=__a			,			encoder_hidden_states=__a			,			added_cond_kwargs=__a			,			return_dict=__a			,			)[0]

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

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

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

     if output_type == "pil":
      SCREAMING_SNAKE_CASE__: Dict=   self.numpy_to_pil(__a )

     if not return_dict:
      return (image,)

     return ImagePipelineOutput(images=__a )





import numpy as np
import torch
import torch.nn as nn
from transformers import CLIPConfig, CLIPVisionModelWithProjection, PreTrainedModel

from ...utils import logging


snake_case__				:	int					       =					logging.get_logger(__name__)





class   SCREAMING_SNAKE_CASE_			(a__						):




       '''simple docstring'''



       _a   							=							CLIPConfig

       _a   							=							["CLIPEncoderLayer"]



       def __init__(						self				:				Tuple		,				__a				:				CLIPConfig							)					->Union[str, Any]:
          super().__init__(__a							)

          lowerCamelCase_						:   List[Any]		= CLIPVisionModelWithProjection(config.vision_config							)

          lowerCamelCase_						:   int		= nn.Linear(config.vision_config.projection_dim		,				1							)
          lowerCamelCase_						:   int		= nn.Linear(config.vision_config.projection_dim		,				1							)




       @torch.no_grad()
       def 				_lowerCAmelCase							(						self				:				Dict		,				__a				:				List[Any]		,				__a				:				Tuple		,				__a				:				int=0.5		,				__a				:				Optional[Any]=0.5							)					->Union[str, Any]:
          lowerCamelCase_						:   Dict		= self.vision_model(__a							)[0]

          lowerCamelCase_						:   str		= self.p_head(__a							)
          lowerCamelCase_						:   Union[str, Any]		= nsfw_detected.flatten()
          lowerCamelCase_						:   Tuple		= nsfw_detected > p_threshold
          lowerCamelCase_						:   Dict		= nsfw_detected.tolist()

          if any(__a							):
             logger.warning(
                 """Potential NSFW content was detected in one or more images. A black image will be returned instead."""
                 """ Try again with a different prompt and/or seed."""							)

          for idx, nsfw_detected_ in enumerate(__a							):
             if nsfw_detected_:
                lowerCamelCase_						:   int		= np.zeros(images[idx].shape							)

          lowerCamelCase_						:   Union[str, Any]		= self.w_head(__a							)
          lowerCamelCase_						:   List[str]		= watermark_detected.flatten()
          lowerCamelCase_						:   Dict		= watermark_detected > w_threshold
          lowerCamelCase_						:   Optional[int]		= watermark_detected.tolist()

          if any(__a							):
             logger.warning(
                 """Potential watermarked content was detected in one or more images. A black image will be returned instead."""
                 """ Try again with a different prompt and/or seed."""							)

          for idx, watermark_detected_ in enumerate(__a							):
             if watermark_detected_:
                lowerCamelCase_						:   Tuple		= np.zeros(images[idx].shape							)

          return images, nsfw_detected, watermark_detected





