




import numpy as np




def 						SCREAMING_SNAKE_CASE__ (   lowercase					)  ->       np.ndarray:
   return 1 / (1 + np.exp(-vector					))




def 						SCREAMING_SNAKE_CASE__ (   lowercase					)  ->       np.ndarray:
   return vector * sigmoid(lowercase					)


if __name__ == "__main__":
       import doctest

       doctest.testmod()











import inspect
from typing import Optional, Union

import numpy as np
import PIL
import torch
from torch.nn import functional as F
from torchvision import transforms
from transformers import CLIPFeatureExtractor, CLIPModel, CLIPTextModel, CLIPTokenizer

from diffusers import (
    AutoencoderKL,
    DDIMScheduler,
    DiffusionPipeline,
    DPMSolverMultistepScheduler,
    LMSDiscreteScheduler,
    PNDMScheduler,
    UNetaDConditionModel,
)
from diffusers.pipelines.stable_diffusion.pipeline_stable_diffusion import StableDiffusionPipelineOutput
from diffusers.utils import (
    PIL_INTERPOLATION,
    randn_tensor,
)




def 						SCREAMING_SNAKE_CASE__ (   lowercase						,lowercase						,lowercase					)  ->       Union[str, Any]:
   if isinstance(lowercase						,torch.Tensor					):
      return image
   elif isinstance(lowercase						,PIL.Image.Image					):
      snake_case				:    str           =	[image]

   if isinstance(image[0]						,PIL.Image.Image					):
      snake_case				:    List[Any]           =	[np.array(i.resize((w, h)						,resample=PIL_INTERPOLATION["""lanczos"""]					)					)[None, :] for i in image]
      snake_case				:    Optional[int]           =	np.concatenate(lowercase						,axis=0					)
      snake_case				:    str           =	np.array(lowercase					).astype(np.floataa					) / 255.0
      snake_case				:    List[str]           =	image.transpose(0						,3						,1						,2					)
      snake_case				:    Any           =	2.0 * image - 1.0
      snake_case				:    Optional[Any]           =	torch.from_numpy(lowercase					)
   elif isinstance(image[0]						,torch.Tensor					):
      snake_case				:    Optional[int]           =	torch.cat(lowercase						,dim=0					)
   return image




def 						SCREAMING_SNAKE_CASE__ (   lowercase						,lowercase						,lowercase						,lowercase=0.9995					)  ->       Optional[int]:
   if not isinstance(lowercase						,np.ndarray					):
      snake_case				:    Any           =	True
      snake_case				:    str           =	va.device
      snake_case				:    Optional[Any]           =	va.cpu().numpy()
      snake_case				:    str           =	va.cpu().numpy()

   snake_case				:    Tuple           =	np.sum(va * va / (np.linalg.norm(lowercase					) * np.linalg.norm(lowercase					))					)
   if np.abs(lowercase					) > DOT_THRESHOLD:
      snake_case				:    Optional[int]           =	(1 - t) * va + t * va
   else:
      snake_case				:    List[Any]           =	np.arccos(lowercase					)
      snake_case				:    str           =	np.sin(lowercase					)
      snake_case				:    int           =	theta_a * t
      snake_case				:    Dict           =	np.sin(lowercase					)
      snake_case				:    Optional[Any]           =	np.sin(theta_a - theta_t					) / sin_theta_a
      snake_case				:    Union[str, Any]           =	sin_theta_t / sin_theta_a
      snake_case				:    Union[str, Any]           =	sa * va + sa * va

   if inputs_are_torch:
      snake_case				:    List[Any]           =	torch.from_numpy(lowercase					).to(lowercase					)

   return va




def 						SCREAMING_SNAKE_CASE__ (   lowercase						,lowercase					)  ->       Tuple:
   snake_case				:    Dict           =	F.normalize(lowercase						,dim=-1					)
   snake_case				:    Optional[Any]           =	F.normalize(lowercase						,dim=-1					)
   return (x - y).norm(dim=-1					).div(2					).arcsin().pow(2					).mul(2					)




def 						SCREAMING_SNAKE_CASE__ (   lowercase						,lowercase					)  ->       str:
   for param in model.parameters():
      snake_case				:    Tuple           =	value




class 						__lowercase		(UpperCamelCase__							):




      """simple docstring"""





      def __init__(				self    ,				A    ,				A    ,				A    ,				A    ,				A    ,				A    ,				A    ,				A=None    ,				A=None    ,				A=None    ,				)		->    List[Any]:
         super().__init__()
         self.register_modules(
             vae=A    ,				text_encoder=A    ,				clip_model=A    ,				tokenizer=A    ,				unet=A    ,				scheduler=A    ,				feature_extractor=A    ,				coca_model=A    ,				coca_tokenizer=A    ,				coca_transform=A    ,				)
         snake_case				:    Optional[int]           =	(
             feature_extractor.size
             if isinstance(feature_extractor.size    ,				A     )
             else feature_extractor.size["""shortest_edge"""]
         )
         snake_case				:    Union[str, Any]           =	transforms.Normalize(mean=feature_extractor.image_mean    ,				std=feature_extractor.image_std     )
         set_requires_grad(self.text_encoder    ,				A     )
         set_requires_grad(self.clip_model    ,				A     )

      def      UpperCAmelCase				(				self    ,				A = "auto"     )		->    Tuple:
         if slice_size == "auto":
            # half the attention head size is usually a good trade-off between
            # speed and memory
            snake_case				:    List[str]           =	self.unet.config.attention_head_dim // 2
         self.unet.set_attention_slice(A     )

      def      UpperCAmelCase				(				self     )		->    Optional[int]:
         self.enable_attention_slicing(A     )

      def      UpperCAmelCase				(				self     )		->    Any:
         set_requires_grad(self.vae    ,				A     )

      def      UpperCAmelCase				(				self     )		->    List[Any]:
         set_requires_grad(self.vae    ,				A     )

      def      UpperCAmelCase				(				self     )		->    Union[str, Any]:
         set_requires_grad(self.unet    ,				A     )

      def      UpperCAmelCase				(				self     )		->    Tuple:
         set_requires_grad(self.unet    ,				A     )

      def      UpperCAmelCase				(				self    ,				A    ,				A    ,				A     )		->    Dict:
         # get the original timestep using init_timestep
         snake_case				:    Tuple           =	min(int(num_inference_steps * strength     )    ,				A     )

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

         return timesteps, num_inference_steps - t_start

      def      UpperCAmelCase				(				self    ,				A    ,				A    ,				A    ,				A    ,				A    ,				A=None     )		->    List[str]:
         if not isinstance(A    ,				torch.Tensor     ):
            raise ValueError(f"""`image` has to be of type `torch.Tensor` but is {type(A     )}"""     )

         snake_case				:    str           =	image.to(device=A    ,				dtype=A     )

         if isinstance(A    ,				A     ):
            snake_case				:    int           =	[
                self.vae.encode(image[i : i + 1]     ).latent_dist.sample(generator[i]     ) for i in range(A     )
            ]
            snake_case				:    str           =	torch.cat(A    ,				dim=0     )
         else:
            snake_case				:    List[Any]           =	self.vae.encode(A     ).latent_dist.sample(A     )

         # Hardcode 0.18215 because stable-diffusion-2-base has not self.vae.config.scaling_factor
         snake_case				:    Dict           =	0.1_82_15 * init_latents
         snake_case				:    Tuple           =	init_latents.repeat_interleave(A    ,				dim=0     )

         snake_case				:    Optional[int]           =	randn_tensor(init_latents.shape    ,				generator=A    ,				device=A    ,				dtype=A     )

         # get latents
         snake_case				:    Union[str, Any]           =	self.scheduler.add_noise(A    ,				A    ,				A     )
         snake_case				:    List[Any]           =	init_latents

         return latents

      def      UpperCAmelCase				(				self    ,				A     )		->    int:
         snake_case				:    Optional[Any]           =	self.coca_transform(A     ).unsqueeze(0     )
         with torch.no_grad(), torch.cuda.amp.autocast():
            snake_case				:    Optional[Any]           =	self.coca_model.generate(transformed_image.to(device=self.device    ,				dtype=self.coca_model.dtype     )     )
         snake_case				:    int           =	self.coca_tokenizer.decode(generated[0].cpu().numpy()     )
         return generated.split("""<end_of_text>"""     )[0].replace("""<start_of_text>"""    ,				""""""     ).rstrip(""" .,"""     )

      def      UpperCAmelCase				(				self    ,				A    ,				A     )		->    List[Any]:
         snake_case				:    Tuple           =	self.feature_extractor.preprocess(A     )
         snake_case				:    List[Any]           =	torch.from_numpy(clip_image_input["""pixel_values"""][0]     ).unsqueeze(0     ).to(self.device     ).half()
         snake_case				:    Optional[int]           =	self.clip_model.get_image_features(A     )
         snake_case				:    List[str]           =	image_embeddings_clip / image_embeddings_clip.norm(p=2    ,				dim=-1    ,				keepdim=A     )
         snake_case				:    Tuple           =	image_embeddings_clip.repeat_interleave(A    ,				dim=0     )
         return image_embeddings_clip

      @torch.enable_grad()
      def      UpperCAmelCase				(				self    ,				A    ,				A    ,				A    ,				A    ,				A    ,				A    ,				A    ,				)		->    Any:
         snake_case				:    Dict           =	latents.detach().requires_grad_()

         snake_case				:    str           =	self.scheduler.scale_model_input(A    ,				A     )

         # predict the noise residual
         snake_case				:    str           =	self.unet(A    ,				A    ,				encoder_hidden_states=A     ).sample

         if isinstance(self.scheduler    ,				(PNDMScheduler, DDIMScheduler, DPMSolverMultistepScheduler)     ):
            snake_case				:    int           =	self.scheduler.alphas_cumprod[timestep]
            snake_case				:    Tuple           =	1 - alpha_prod_t
            # compute predicted original sample from predicted noise also called
            # "predicted x_0" of formula (12) from https://arxiv.org/pdf/2010.02502.pdf
            snake_case				:    Any           =	(latents - beta_prod_t ** 0.5 * noise_pred) / alpha_prod_t ** 0.5

            snake_case				:    str           =	torch.sqrt(A     )
            snake_case				:    str           =	pred_original_sample * (fac) + latents * (1 - fac)
         elif isinstance(self.scheduler    ,				A     ):
            snake_case				:    int           =	self.scheduler.sigmas[index]
            snake_case				:    List[Any]           =	latents - sigma * noise_pred
         else:
            raise ValueError(f"""scheduler type {type(self.scheduler     )} not supported"""     )

         # Hardcode 0.18215 because stable-diffusion-2-base has not self.vae.config.scaling_factor
         snake_case				:    List[str]           =	1 / 0.1_82_15 * sample
         snake_case				:    str           =	self.vae.decode(A     ).sample
         snake_case				:    Any           =	(image / 2 + 0.5).clamp(0    ,				1     )

         snake_case				:    str           =	transforms.Resize(self.feature_extractor_size     )(A     )
         snake_case				:    Dict           =	self.normalize(A     ).to(latents.dtype     )

         snake_case				:    Union[str, Any]           =	self.clip_model.get_image_features(A     )
         snake_case				:    List[str]           =	image_embeddings_clip / image_embeddings_clip.norm(p=2    ,				dim=-1    ,				keepdim=A     )

         snake_case				:    Optional[int]           =	spherical_dist_loss(A    ,				A     ).mean() * clip_guidance_scale

         snake_case				:    int           =	-torch.autograd.grad(A    ,				A     )[0]

         if isinstance(self.scheduler    ,				A     ):
            snake_case				:    Union[str, Any]           =	latents.detach() + grads * (sigma**2)
            snake_case				:    Union[str, Any]           =	noise_pred_original
         else:
            snake_case				:    List[str]           =	noise_pred_original - torch.sqrt(A     ) * grads
         return noise_pred, latents

      @torch.no_grad()
      def __call__(				self    ,				A    ,				A    ,				A = None    ,				A = None    ,				A = 5_1_2    ,				A = 5_1_2    ,				A = 0.6    ,				A = 5_0    ,				A = 7.5    ,				A = 1    ,				A = 0.0    ,				A = 1_0_0    ,				A = None    ,				A = "pil"    ,				A = True    ,				A = 0.8    ,				A = 0.1    ,				A = 0.1    ,				)		->    Union[str, Any]:
         if isinstance(A    ,				A     ) and len(A     ) != batch_size:
            raise ValueError(f"""You have passed {batch_size} batch_size, but only {len(A     )} generators."""     )

         if height % 8 != 0 or width % 8 != 0:
            raise ValueError(f"""`height` and `width` have to be divisible by 8 but are {height} and {width}."""     )

         if isinstance(A    ,				torch.Generator     ) and batch_size > 1:
            snake_case				:    Dict           =	[generator] + [None] * (batch_size - 1)

         snake_case				:    Tuple           =	[
             ("""model""", self.coca_model is None),
             ("""tokenizer""", self.coca_tokenizer is None),
             ("""transform""", self.coca_transform is None),
         ]
         snake_case				:    List[str]           =	[x[0] for x in coca_is_none if x[1]]
         snake_case				:    Optional[int]           =	""", """.join(A     )
         # generate prompts with coca model if prompt is None
         if content_prompt is None:
            if len(A     ):
               raise ValueError(
                   f"""Content prompt is None and CoCa [{coca_is_none_str}] is None."""
                   f"""Set prompt or pass Coca [{coca_is_none_str}] to DiffusionPipeline."""     )
            snake_case				:    Tuple           =	self.get_image_description(A     )
         if style_prompt is None:
            if len(A     ):
               raise ValueError(
                   f"""Style prompt is None and CoCa [{coca_is_none_str}] is None."""
                   f""" Set prompt or pass Coca [{coca_is_none_str}] to DiffusionPipeline."""     )
            snake_case				:    List[Any]           =	self.get_image_description(A     )

         # get prompt text embeddings for content and style
         snake_case				:    Dict           =	self.tokenizer(
             A    ,				padding="""max_length"""    ,				max_length=self.tokenizer.model_max_length    ,				truncation=A    ,				return_tensors="""pt"""    ,				)
         snake_case				:    str           =	self.text_encoder(content_text_input.input_ids.to(self.device     )     )[0]

         snake_case				:    Dict           =	self.tokenizer(
             A    ,				padding="""max_length"""    ,				max_length=self.tokenizer.model_max_length    ,				truncation=A    ,				return_tensors="""pt"""    ,				)
         snake_case				:    Tuple           =	self.text_encoder(style_text_input.input_ids.to(self.device     )     )[0]

         snake_case				:    List[str]           =	slerp(A    ,				A    ,				A     )

         # duplicate text embeddings for each generation per prompt
         snake_case				:    List[Any]           =	text_embeddings.repeat_interleave(A    ,				dim=0     )

         # set timesteps
         snake_case				:    Union[str, Any]           =	"""offset""" in set(inspect.signature(self.scheduler.set_timesteps     ).parameters.keys()     )
         snake_case				:    Optional[Any]           =	{}
         if accepts_offset:
            snake_case				:    Dict           =	1

         self.scheduler.set_timesteps(A    ,				**A     )
         # Some schedulers like PNDM have timesteps as arrays
         # It's more optimized to move all timesteps to correct device beforehand
         self.scheduler.timesteps.to(self.device     )

         snake_case     , snake_case				:    List[Any]           =	self.get_timesteps(A    ,				A    ,				self.device     )
         snake_case				:    List[str]           =	timesteps[:1].repeat(A     )

         # Preprocess image
         snake_case				:    Dict           =	preprocess(A    ,				A    ,				A     )
         snake_case				:    List[Any]           =	self.prepare_latents(
             A    ,				A    ,				A    ,				text_embeddings.dtype    ,				self.device    ,				A     )

         snake_case				:    Optional[int]           =	preprocess(A    ,				A    ,				A     )
         snake_case				:    Optional[Any]           =	self.prepare_latents(
             A    ,				A    ,				A    ,				text_embeddings.dtype    ,				self.device    ,				A     )

         snake_case				:    str           =	slerp(A    ,				A    ,				A     )

         if clip_guidance_scale > 0:
            snake_case				:    List[Any]           =	self.get_clip_image_embeddings(A    ,				A     )
            snake_case				:    Any           =	self.get_clip_image_embeddings(A    ,				A     )
            snake_case				:    Tuple           =	slerp(
                A    ,				A    ,				A     )

         # here `guidance_scale` is defined analog to the guidance weight `w` of equation (2)
         # of the Imagen paper: https://arxiv.org/pdf/2205.11487.pdf . `guidance_scale = 1`
         # corresponds to doing no classifier free guidance.
         snake_case				:    Optional[Any]           =	guidance_scale > 1.0
         # get unconditional embeddings for classifier free guidance
         if do_classifier_free_guidance:
            snake_case				:    List[str]           =	content_text_input.input_ids.shape[-1]
            snake_case				:    Any           =	self.tokenizer([""""""]    ,				padding="""max_length"""    ,				max_length=A    ,				return_tensors="""pt"""     )
            snake_case				:    Tuple           =	self.text_encoder(uncond_input.input_ids.to(self.device     )     )[0]
            # duplicate unconditional embeddings for each generation per prompt
            snake_case				:    Tuple           =	uncond_embeddings.repeat_interleave(A    ,				dim=0     )

            # For classifier free guidance, we need to do two forward passes.
            # Here we concatenate the unconditional and text embeddings into a single batch
            # to avoid doing two forward passes
            snake_case				:    str           =	torch.cat([uncond_embeddings, text_embeddings]     )

         # get the initial random noise unless the user supplied it

         # Unlike in other pipelines, latents need to be generated in the target device
         # for 1-to-1 results reproducibility with the CompVis implementation.
         # However this currently doesn't work in `mps`.
         snake_case				:    Tuple           =	(batch_size, self.unet.config.in_channels, height // 8, width // 8)
         snake_case				:    List[Any]           =	text_embeddings.dtype
         if latents is None:
            if self.device.type == "mps":
               # randn does not work reproducibly on mps
               snake_case				:    List[Any]           =	torch.randn(A    ,				generator=A    ,				device="""cpu"""    ,				dtype=A     ).to(
                   self.device     )
            else:
               snake_case				:    Optional[int]           =	torch.randn(A    ,				generator=A    ,				device=self.device    ,				dtype=A     )
         else:
            if latents.shape != latents_shape:
               raise ValueError(f"""Unexpected latents shape, got {latents.shape}, expected {latents_shape}"""     )
            snake_case				:    Union[str, Any]           =	latents.to(self.device     )

         # scale the initial noise by the standard deviation required by the scheduler
         snake_case				:    Dict           =	latents * self.scheduler.init_noise_sigma

         # prepare extra kwargs for the scheduler step, since not all schedulers have the same signature
         # eta (η) is only used with the DDIMScheduler, it will be ignored for other schedulers.
         # eta corresponds to η in DDIM paper: https://arxiv.org/abs/2010.02502
         # and should be between [0, 1]
         snake_case				:    Tuple           =	"""eta""" in set(inspect.signature(self.scheduler.step     ).parameters.keys()     )
         snake_case				:    Dict           =	{}
         if accepts_eta:
            snake_case				:    Union[str, Any]           =	eta

         # check if the scheduler accepts generator
         snake_case				:    List[Any]           =	"""generator""" in set(inspect.signature(self.scheduler.step     ).parameters.keys()     )
         if accepts_generator:
            snake_case				:    List[str]           =	generator

         with self.progress_bar(total=A     ):
            for i, t in enumerate(A     ):
               # expand the latents if we are doing classifier free guidance
               snake_case				:    Optional[Any]           =	torch.cat([latents] * 2     ) if do_classifier_free_guidance else latents
               snake_case				:    List[str]           =	self.scheduler.scale_model_input(A    ,				A     )

               # predict the noise residual
               snake_case				:    Any           =	self.unet(A    ,				A    ,				encoder_hidden_states=A     ).sample

               # perform classifier free guidance
               if do_classifier_free_guidance:
                  snake_case     , snake_case				:    int           =	noise_pred.chunk(2     )
                  snake_case				:    Tuple           =	noise_pred_uncond + guidance_scale * (noise_pred_text - noise_pred_uncond)

               # perform clip guidance
               if clip_guidance_scale > 0:
                  snake_case				:    Any           =	(
                      text_embeddings.chunk(2     )[1] if do_classifier_free_guidance else text_embeddings
                  )
                  snake_case     , snake_case				:    List[Any]           =	self.cond_fn(
                      A    ,				A    ,				A    ,				A    ,				A    ,				A    ,				A    ,				)

               # compute the previous noisy sample x_t -> x_t-1
               snake_case				:    Tuple           =	self.scheduler.step(A    ,				A    ,				A    ,				**A     ).prev_sample

        # Hardcode 0.18215 because stable-diffusion-2-base has not self.vae.config.scaling_factor
         snake_case				:    str           =	1 / 0.1_82_15 * latents
         snake_case				:    Optional[Any]           =	self.vae.decode(A     ).sample

         snake_case				:    Union[str, Any]           =	(image / 2 + 0.5).clamp(0    ,				1     )
         snake_case				:    str           =	image.cpu().permute(0    ,				2    ,				3    ,				1     ).numpy()

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

         if not return_dict:
            return (image, None)

         return StableDiffusionPipelineOutput(images=A    ,				nsfw_content_detected=A     )





