




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 					UpperCAmelCase_							(      __UpperCAmelCase						:	Optional[Any]			,			__UpperCAmelCase						:	int			,			__UpperCAmelCase						:	Optional[int]							)					-> Union[str, Any]:
      if isinstance(__UpperCAmelCase			,			torch.Tensor							):
            return image
      elif isinstance(__UpperCAmelCase			,			PIL.Image.Image							):
            SCREAMING_SNAKE_CASE_							=			[image]

      if isinstance(image[0]			,			PIL.Image.Image							):
            SCREAMING_SNAKE_CASE_							=			[np.array(i.resize((w, h)			,			resample=PIL_INTERPOLATION['lanczos']							)							)[None, :] for i in image]
            SCREAMING_SNAKE_CASE_							=			np.concatenate(__UpperCAmelCase			,			axis=0							)
            SCREAMING_SNAKE_CASE_							=			np.array(__UpperCAmelCase							).astype(np.floataa							) / 2_5_5.0
            SCREAMING_SNAKE_CASE_							=			image.transpose(0			,			3			,			1			,			2							)
            SCREAMING_SNAKE_CASE_							=			2.0 * image - 1.0
            SCREAMING_SNAKE_CASE_							=			torch.from_numpy(__UpperCAmelCase							)
      elif isinstance(image[0]			,			torch.Tensor							):
            SCREAMING_SNAKE_CASE_							=			torch.cat(__UpperCAmelCase			,			dim=0							)
      return image

def 					UpperCAmelCase_							(      __UpperCAmelCase						:	Tuple			,			__UpperCAmelCase						:	str			,			__UpperCAmelCase						:	Dict			,			__UpperCAmelCase						:	int=0.9_9_9_5							)					-> List[Any]:
      if not isinstance(__UpperCAmelCase			,			np.ndarray							):
            SCREAMING_SNAKE_CASE_							=			True
            SCREAMING_SNAKE_CASE_							=			va.device
            SCREAMING_SNAKE_CASE_							=			va.cpu().numpy()
            SCREAMING_SNAKE_CASE_							=			va.cpu().numpy()

      SCREAMING_SNAKE_CASE_							=			np.sum(va * va / (np.linalg.norm(__UpperCAmelCase							) * np.linalg.norm(__UpperCAmelCase							))							)
      if np.abs(__UpperCAmelCase							) > DOT_THRESHOLD:
            SCREAMING_SNAKE_CASE_							=			(1 - t) * va + t * va
      else:
            SCREAMING_SNAKE_CASE_							=			np.arccos(__UpperCAmelCase							)
            SCREAMING_SNAKE_CASE_							=			np.sin(__UpperCAmelCase							)
            SCREAMING_SNAKE_CASE_							=			theta_a * t
            SCREAMING_SNAKE_CASE_							=			np.sin(__UpperCAmelCase							)
            SCREAMING_SNAKE_CASE_							=			np.sin(theta_a - theta_t							) / sin_theta_a
            SCREAMING_SNAKE_CASE_							=			sin_theta_t / sin_theta_a
            SCREAMING_SNAKE_CASE_							=			sa * va + sa * va

      if inputs_are_torch:
            SCREAMING_SNAKE_CASE_							=			torch.from_numpy(__UpperCAmelCase							).to(__UpperCAmelCase							)

      return va

def 					UpperCAmelCase_							(      __UpperCAmelCase						:	str			,			__UpperCAmelCase						:	List[str]							)					-> Union[str, Any]:
      SCREAMING_SNAKE_CASE_							=			F.normalize(__UpperCAmelCase			,			dim=-1							)
      SCREAMING_SNAKE_CASE_							=			F.normalize(__UpperCAmelCase			,			dim=-1							)
      return (x - y).norm(dim=-1							).div(2							).arcsin().pow(2							).mul(2							)

def 					UpperCAmelCase_							(      __UpperCAmelCase						:	Dict			,			__UpperCAmelCase						:	Optional[int]							)					-> int:
      for param in model.parameters():
            SCREAMING_SNAKE_CASE_							=			value





class        lowerCamelCase_							(      _SCREAMING_SNAKE_CASE					):


 '''simple docstring'''







 def __init__(							self  :		Any				,				_lowerCAmelCase  :		AutoencoderKL				,				_lowerCAmelCase  :		CLIPTextModel				,				_lowerCAmelCase  :		CLIPModel				,				_lowerCAmelCase  :		CLIPTokenizer				,				_lowerCAmelCase  :		UNetaDConditionModel				,				_lowerCAmelCase  :		Union[PNDMScheduler, LMSDiscreteScheduler, DDIMScheduler, DPMSolverMultistepScheduler]				,				_lowerCAmelCase  :		CLIPFeatureExtractor				,				_lowerCAmelCase  :		Tuple=None				,				_lowerCAmelCase  :		Union[str, Any]=None				,				_lowerCAmelCase  :		Union[str, Any]=None				,				):
       super().__init__()
       self.register_modules(
           vae=_lowerCAmelCase				,				text_encoder=_lowerCAmelCase				,				clip_model=_lowerCAmelCase				,				tokenizer=_lowerCAmelCase				,				unet=_lowerCAmelCase				,				scheduler=_lowerCAmelCase				,				feature_extractor=_lowerCAmelCase				,				coca_model=_lowerCAmelCase				,				coca_tokenizer=_lowerCAmelCase				,				coca_transform=_lowerCAmelCase				,				)
       SCREAMING_SNAKE_CASE_							=			(
           feature_extractor.size
           if isinstance(feature_extractor.size				,				_lowerCAmelCase						)
           else feature_extractor.size['shortest_edge']
       )
       SCREAMING_SNAKE_CASE_							=			transforms.Normalize(mean=feature_extractor.image_mean				,				std=feature_extractor.image_std						)
       set_requires_grad(self.text_encoder				,				_lowerCAmelCase						)
       set_requires_grad(self.clip_model				,				_lowerCAmelCase						)



 def       lowerCAmelCase_  (							self  :		Union[str, Any]				,				_lowerCAmelCase  :		Optional[Union[str, int]] = "auto"						):
       if slice_size == "auto":
             # half the attention head size is usually a good trade-off between
             # speed and memory
             SCREAMING_SNAKE_CASE_							=			self.unet.config.attention_head_dim // 2
       self.unet.set_attention_slice(_lowerCAmelCase						)



 def       lowerCAmelCase_  (							self  :		List[Any]						):
       self.enable_attention_slicing(_lowerCAmelCase						)



 def       lowerCAmelCase_  (							self  :		Optional[Any]						):
       set_requires_grad(self.vae				,				_lowerCAmelCase						)



 def       lowerCAmelCase_  (							self  :		Optional[int]						):
       set_requires_grad(self.vae				,				_lowerCAmelCase						)



 def       lowerCAmelCase_  (							self  :		int						):
       set_requires_grad(self.unet				,				_lowerCAmelCase						)



 def       lowerCAmelCase_  (							self  :		List[str]						):
       set_requires_grad(self.unet				,				_lowerCAmelCase						)



 def       lowerCAmelCase_  (							self  :		Optional[int]				,				_lowerCAmelCase  :		Union[str, Any]				,				_lowerCAmelCase  :		Optional[int]				,				_lowerCAmelCase  :		Any						):
       # get the original timestep using init_timestep
       SCREAMING_SNAKE_CASE_							=			min(int(num_inference_steps * strength						)				,				_lowerCAmelCase						)

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

       return timesteps, num_inference_steps - t_start



 def       lowerCAmelCase_  (							self  :		Union[str, Any]				,				_lowerCAmelCase  :		Optional[Any]				,				_lowerCAmelCase  :		List[Any]				,				_lowerCAmelCase  :		Optional[Any]				,				_lowerCAmelCase  :		int				,				_lowerCAmelCase  :		Any				,				_lowerCAmelCase  :		Optional[Any]=None						):
       if not isinstance(_lowerCAmelCase				,				torch.Tensor						):
             raise ValueError(F"`image` has to be of type `torch.Tensor` but is {type(_lowerCAmelCase						)}"						)

       SCREAMING_SNAKE_CASE_							=			image.to(device=_lowerCAmelCase				,				dtype=_lowerCAmelCase						)

       if isinstance(_lowerCAmelCase				,				_lowerCAmelCase						):
             SCREAMING_SNAKE_CASE_							=			[
                 self.vae.encode(image[i : i + 1]						).latent_dist.sample(generator[i]						) for i in range(_lowerCAmelCase						)
             ]
             SCREAMING_SNAKE_CASE_							=			torch.cat(_lowerCAmelCase				,				dim=0						)
       else:
             SCREAMING_SNAKE_CASE_							=			self.vae.encode(_lowerCAmelCase						).latent_dist.sample(_lowerCAmelCase						)

       # Hardcode 0.18215 because stable-diffusion-2-base has not self.vae.config.scaling_factor
       SCREAMING_SNAKE_CASE_							=			0.1_8215 * init_latents
       SCREAMING_SNAKE_CASE_							=			init_latents.repeat_interleave(_lowerCAmelCase				,				dim=0						)

       SCREAMING_SNAKE_CASE_							=			randn_tensor(init_latents.shape				,				generator=_lowerCAmelCase				,				device=_lowerCAmelCase				,				dtype=_lowerCAmelCase						)

       # get latents
       SCREAMING_SNAKE_CASE_							=			self.scheduler.add_noise(_lowerCAmelCase				,				_lowerCAmelCase				,				_lowerCAmelCase						)
       SCREAMING_SNAKE_CASE_							=			init_latents

       return latents



 def       lowerCAmelCase_  (							self  :		Tuple				,				_lowerCAmelCase  :		int						):
       SCREAMING_SNAKE_CASE_							=			self.coca_transform(_lowerCAmelCase						).unsqueeze(0						)
       with torch.no_grad(), torch.cuda.amp.autocast():
             SCREAMING_SNAKE_CASE_							=			self.coca_model.generate(transformed_image.to(device=self.device				,				dtype=self.coca_model.dtype						)						)
       SCREAMING_SNAKE_CASE_							=			self.coca_tokenizer.decode(generated[0].cpu().numpy()						)
       return generated.split('<end_of_text>'						)[0].replace('<start_of_text>'				,				''						).rstrip(' .,'						)



 def       lowerCAmelCase_  (							self  :		Dict				,				_lowerCAmelCase  :		Optional[int]				,				_lowerCAmelCase  :		List[str]						):
       SCREAMING_SNAKE_CASE_							=			self.feature_extractor.preprocess(_lowerCAmelCase						)
       SCREAMING_SNAKE_CASE_							=			torch.from_numpy(clip_image_input['pixel_values'][0]						).unsqueeze(0						).to(self.device						).half()
       SCREAMING_SNAKE_CASE_							=			self.clip_model.get_image_features(_lowerCAmelCase						)
       SCREAMING_SNAKE_CASE_							=			image_embeddings_clip / image_embeddings_clip.norm(p=2				,				dim=-1				,				keepdim=_lowerCAmelCase						)
       SCREAMING_SNAKE_CASE_							=			image_embeddings_clip.repeat_interleave(_lowerCAmelCase				,				dim=0						)
       return image_embeddings_clip



 @torch.enable_grad()
 def       lowerCAmelCase_  (							self  :		Optional[Any]				,				_lowerCAmelCase  :		str				,				_lowerCAmelCase  :		int				,				_lowerCAmelCase  :		Union[str, Any]				,				_lowerCAmelCase  :		int				,				_lowerCAmelCase  :		Optional[int]				,				_lowerCAmelCase  :		Optional[int]				,				_lowerCAmelCase  :		List[str]				,				):
       SCREAMING_SNAKE_CASE_							=			latents.detach().requires_grad_()

       SCREAMING_SNAKE_CASE_							=			self.scheduler.scale_model_input(_lowerCAmelCase				,				_lowerCAmelCase						)

       # predict the noise residual
       SCREAMING_SNAKE_CASE_							=			self.unet(_lowerCAmelCase				,				_lowerCAmelCase				,				encoder_hidden_states=_lowerCAmelCase						).sample

       if isinstance(self.scheduler				,				(PNDMScheduler, DDIMScheduler, DPMSolverMultistepScheduler)						):
             SCREAMING_SNAKE_CASE_							=			self.scheduler.alphas_cumprod[timestep]
             SCREAMING_SNAKE_CASE_							=			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
             SCREAMING_SNAKE_CASE_							=			(latents - beta_prod_t ** 0.5 * noise_pred) / alpha_prod_t ** 0.5

             SCREAMING_SNAKE_CASE_							=			torch.sqrt(_lowerCAmelCase						)
             SCREAMING_SNAKE_CASE_							=			pred_original_sample * (fac) + latents * (1 - fac)
       elif isinstance(self.scheduler				,				_lowerCAmelCase						):
             SCREAMING_SNAKE_CASE_							=			self.scheduler.sigmas[index]
             SCREAMING_SNAKE_CASE_							=			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
       SCREAMING_SNAKE_CASE_							=			1 / 0.1_8215 * sample
       SCREAMING_SNAKE_CASE_							=			self.vae.decode(_lowerCAmelCase						).sample
       SCREAMING_SNAKE_CASE_							=			(image / 2 + 0.5).clamp(0				,				1						)

       SCREAMING_SNAKE_CASE_							=			transforms.Resize(self.feature_extractor_size						)(_lowerCAmelCase						)
       SCREAMING_SNAKE_CASE_							=			self.normalize(_lowerCAmelCase						).to(latents.dtype						)

       SCREAMING_SNAKE_CASE_							=			self.clip_model.get_image_features(_lowerCAmelCase						)
       SCREAMING_SNAKE_CASE_							=			image_embeddings_clip / image_embeddings_clip.norm(p=2				,				dim=-1				,				keepdim=_lowerCAmelCase						)

       SCREAMING_SNAKE_CASE_							=			spherical_dist_loss(_lowerCAmelCase				,				_lowerCAmelCase						).mean() * clip_guidance_scale

       SCREAMING_SNAKE_CASE_							=			-torch.autograd.grad(_lowerCAmelCase				,				_lowerCAmelCase						)[0]

       if isinstance(self.scheduler				,				_lowerCAmelCase						):
             SCREAMING_SNAKE_CASE_							=			latents.detach() + grads * (sigma**2)
             SCREAMING_SNAKE_CASE_							=			noise_pred_original
       else:
             SCREAMING_SNAKE_CASE_							=			noise_pred_original - torch.sqrt(_lowerCAmelCase						) * grads
       return noise_pred, latents




 @torch.no_grad()
 def __call__(							self  :		int				,				_lowerCAmelCase  :		Union[torch.FloatTensor, PIL.Image.Image]				,				_lowerCAmelCase  :		Union[torch.FloatTensor, PIL.Image.Image]				,				_lowerCAmelCase  :		Optional[str] = None				,				_lowerCAmelCase  :		Optional[str] = None				,				_lowerCAmelCase  :		Optional[int] = 512				,				_lowerCAmelCase  :		Optional[int] = 512				,				_lowerCAmelCase  :		float = 0.6				,				_lowerCAmelCase  :		Optional[int] = 50				,				_lowerCAmelCase  :		Optional[float] = 7.5				,				_lowerCAmelCase  :		Optional[int] = 1				,				_lowerCAmelCase  :		float = 0.0				,				_lowerCAmelCase  :		Optional[float] = 100				,				_lowerCAmelCase  :		Optional[torch.Generator] = None				,				_lowerCAmelCase  :		Optional[str] = "pil"				,				_lowerCAmelCase  :		bool = True				,				_lowerCAmelCase  :		float = 0.8				,				_lowerCAmelCase  :		float = 0.1				,				_lowerCAmelCase  :		float = 0.1				,				):
       if isinstance(_lowerCAmelCase				,				_lowerCAmelCase						) and len(_lowerCAmelCase						) != batch_size:
             raise ValueError(F"You have passed {batch_size} batch_size, but only {len(_lowerCAmelCase						)} 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(_lowerCAmelCase				,				torch.Generator						) and batch_size > 1:
             SCREAMING_SNAKE_CASE_							=			[generator] + [None] * (batch_size - 1)

       SCREAMING_SNAKE_CASE_							=			[
           ('model', self.coca_model is None),
           ('tokenizer', self.coca_tokenizer is None),
           ('transform', self.coca_transform is None),
       ]
       SCREAMING_SNAKE_CASE_							=			[x[0] for x in coca_is_none if x[1]]
       SCREAMING_SNAKE_CASE_							=			', '.join(_lowerCAmelCase						)
       # generate prompts with coca model if prompt is None
       if content_prompt is None:
             if len(_lowerCAmelCase						):
                   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."						)
             SCREAMING_SNAKE_CASE_							=			self.get_image_description(_lowerCAmelCase						)
       if style_prompt is None:
             if len(_lowerCAmelCase						):
                   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."						)
             SCREAMING_SNAKE_CASE_							=			self.get_image_description(_lowerCAmelCase						)

       # get prompt text embeddings for content and style
       SCREAMING_SNAKE_CASE_							=			self.tokenizer(
           _lowerCAmelCase				,				padding='max_length'				,				max_length=self.tokenizer.model_max_length				,				truncation=_lowerCAmelCase				,				return_tensors='pt'				,				)
       SCREAMING_SNAKE_CASE_							=			self.text_encoder(content_text_input.input_ids.to(self.device						)						)[0]

       SCREAMING_SNAKE_CASE_							=			self.tokenizer(
           _lowerCAmelCase				,				padding='max_length'				,				max_length=self.tokenizer.model_max_length				,				truncation=_lowerCAmelCase				,				return_tensors='pt'				,				)
       SCREAMING_SNAKE_CASE_							=			self.text_encoder(style_text_input.input_ids.to(self.device						)						)[0]

       SCREAMING_SNAKE_CASE_							=			slerp(_lowerCAmelCase				,				_lowerCAmelCase				,				_lowerCAmelCase						)

       # duplicate text embeddings for each generation per prompt
       SCREAMING_SNAKE_CASE_							=			text_embeddings.repeat_interleave(_lowerCAmelCase				,				dim=0						)

       # set timesteps
       SCREAMING_SNAKE_CASE_							=			'offset' in set(inspect.signature(self.scheduler.set_timesteps						).parameters.keys()						)
       SCREAMING_SNAKE_CASE_							=			{}
       if accepts_offset:
             SCREAMING_SNAKE_CASE_							=			1

       self.scheduler.set_timesteps(_lowerCAmelCase				,				**_lowerCAmelCase						)
       # 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						)

       SCREAMING_SNAKE_CASE_						, SCREAMING_SNAKE_CASE_							=			self.get_timesteps(_lowerCAmelCase				,				_lowerCAmelCase				,				self.device						)
       SCREAMING_SNAKE_CASE_							=			timesteps[:1].repeat(_lowerCAmelCase						)

       # Preprocess image
       SCREAMING_SNAKE_CASE_							=			preprocess(_lowerCAmelCase				,				_lowerCAmelCase				,				_lowerCAmelCase						)
       SCREAMING_SNAKE_CASE_							=			self.prepare_latents(
           _lowerCAmelCase				,				_lowerCAmelCase				,				_lowerCAmelCase				,				text_embeddings.dtype				,				self.device				,				_lowerCAmelCase						)

       SCREAMING_SNAKE_CASE_							=			preprocess(_lowerCAmelCase				,				_lowerCAmelCase				,				_lowerCAmelCase						)
       SCREAMING_SNAKE_CASE_							=			self.prepare_latents(
           _lowerCAmelCase				,				_lowerCAmelCase				,				_lowerCAmelCase				,				text_embeddings.dtype				,				self.device				,				_lowerCAmelCase						)

       SCREAMING_SNAKE_CASE_							=			slerp(_lowerCAmelCase				,				_lowerCAmelCase				,				_lowerCAmelCase						)

       if clip_guidance_scale > 0:
             SCREAMING_SNAKE_CASE_							=			self.get_clip_image_embeddings(_lowerCAmelCase				,				_lowerCAmelCase						)
             SCREAMING_SNAKE_CASE_							=			self.get_clip_image_embeddings(_lowerCAmelCase				,				_lowerCAmelCase						)
             SCREAMING_SNAKE_CASE_							=			slerp(
                 _lowerCAmelCase				,				_lowerCAmelCase				,				_lowerCAmelCase						)

       # 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.
       SCREAMING_SNAKE_CASE_							=			guidance_scale > 1.0
       # get unconditional embeddings for classifier free guidance
       if do_classifier_free_guidance:
             SCREAMING_SNAKE_CASE_							=			content_text_input.input_ids.shape[-1]
             SCREAMING_SNAKE_CASE_							=			self.tokenizer(['']				,				padding='max_length'				,				max_length=_lowerCAmelCase				,				return_tensors='pt'						)
             SCREAMING_SNAKE_CASE_							=			self.text_encoder(uncond_input.input_ids.to(self.device						)						)[0]
             # duplicate unconditional embeddings for each generation per prompt
             SCREAMING_SNAKE_CASE_							=			uncond_embeddings.repeat_interleave(_lowerCAmelCase				,				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
             SCREAMING_SNAKE_CASE_							=			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`.
       SCREAMING_SNAKE_CASE_							=			(batch_size, self.unet.config.in_channels, height // 8, width // 8)
       SCREAMING_SNAKE_CASE_							=			text_embeddings.dtype
       if latents is None:
             if self.device.type == "mps":
                   # randn does not work reproducibly on mps
                   SCREAMING_SNAKE_CASE_							=			torch.randn(_lowerCAmelCase				,				generator=_lowerCAmelCase				,				device='cpu'				,				dtype=_lowerCAmelCase						).to(
                       self.device						)
             else:
                   SCREAMING_SNAKE_CASE_							=			torch.randn(_lowerCAmelCase				,				generator=_lowerCAmelCase				,				device=self.device				,				dtype=_lowerCAmelCase						)
       else:
             if latents.shape != latents_shape:
                   raise ValueError(F"Unexpected latents shape, got {latents.shape}, expected {latents_shape}"						)
             SCREAMING_SNAKE_CASE_							=			latents.to(self.device						)

       # scale the initial noise by the standard deviation required by the scheduler
       SCREAMING_SNAKE_CASE_							=			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]
       SCREAMING_SNAKE_CASE_							=			'eta' in set(inspect.signature(self.scheduler.step						).parameters.keys()						)
       SCREAMING_SNAKE_CASE_							=			{}
       if accepts_eta:
             SCREAMING_SNAKE_CASE_							=			eta

       # check if the scheduler accepts generator
       SCREAMING_SNAKE_CASE_							=			'generator' in set(inspect.signature(self.scheduler.step						).parameters.keys()						)
       if accepts_generator:
             SCREAMING_SNAKE_CASE_							=			generator

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

                   # predict the noise residual
                   SCREAMING_SNAKE_CASE_							=			self.unet(_lowerCAmelCase				,				_lowerCAmelCase				,				encoder_hidden_states=_lowerCAmelCase						).sample

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

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

                   # compute the previous noisy sample x_t -> x_t-1
                   SCREAMING_SNAKE_CASE_							=			self.scheduler.step(_lowerCAmelCase				,				_lowerCAmelCase				,				_lowerCAmelCase				,				**_lowerCAmelCase						).prev_sample

        # Hardcode 0.18215 because stable-diffusion-2-base has not self.vae.config.scaling_factor
       SCREAMING_SNAKE_CASE_							=			1 / 0.1_8215 * latents
       SCREAMING_SNAKE_CASE_							=			self.vae.decode(_lowerCAmelCase						).sample

       SCREAMING_SNAKE_CASE_							=			(image / 2 + 0.5).clamp(0				,				1						)
       SCREAMING_SNAKE_CASE_							=			image.cpu().permute(0				,				2				,				3				,				1						).numpy()

       if output_type == "pil":
             SCREAMING_SNAKE_CASE_							=			self.numpy_to_pil(_lowerCAmelCase						)

       if not return_dict:
             return (image, None)

       return StableDiffusionPipelineOutput(images=_lowerCAmelCase				,				nsfw_content_detected=_lowerCAmelCase						)





import argparse
import json

import requests
import torch
from huggingface_hub import hf_hub_download
from PIL import Image

from transformers import ViTImageProcessor, ViTMSNConfig, ViTMSNModel
from transformers.image_utils import IMAGENET_DEFAULT_MEAN, IMAGENET_DEFAULT_STD


torch.set_grad_enabled(False)

def 					UpperCAmelCase_							(      __UpperCAmelCase						:	Any			,			__UpperCAmelCase						:	str=False							)					-> List[str]:
      SCREAMING_SNAKE_CASE_							=			[]
      for i in range(config.num_hidden_layers							):
            # encoder layers: output projection, 2 feedforward neural networks and 2 layernorms
            rename_keys.append((f"module.blocks.{i}.norm1.weight", f"vit.encoder.layer.{i}.layernorm_before.weight")							)
            rename_keys.append((f"module.blocks.{i}.norm1.bias", f"vit.encoder.layer.{i}.layernorm_before.bias")							)
            rename_keys.append(
                (f"module.blocks.{i}.attn.proj.weight", f"vit.encoder.layer.{i}.attention.output.dense.weight")							)
            rename_keys.append((f"module.blocks.{i}.attn.proj.bias", f"vit.encoder.layer.{i}.attention.output.dense.bias")							)
            rename_keys.append((f"module.blocks.{i}.norm2.weight", f"vit.encoder.layer.{i}.layernorm_after.weight")							)
            rename_keys.append((f"module.blocks.{i}.norm2.bias", f"vit.encoder.layer.{i}.layernorm_after.bias")							)
            rename_keys.append((f"module.blocks.{i}.mlp.fc1.weight", f"vit.encoder.layer.{i}.intermediate.dense.weight")							)
            rename_keys.append((f"module.blocks.{i}.mlp.fc1.bias", f"vit.encoder.layer.{i}.intermediate.dense.bias")							)
            rename_keys.append((f"module.blocks.{i}.mlp.fc2.weight", f"vit.encoder.layer.{i}.output.dense.weight")							)
            rename_keys.append((f"module.blocks.{i}.mlp.fc2.bias", f"vit.encoder.layer.{i}.output.dense.bias")							)

      # projection layer + position embeddings
      rename_keys.extend(
          [
              ('module.cls_token', 'vit.embeddings.cls_token'),
              ('module.patch_embed.proj.weight', 'vit.embeddings.patch_embeddings.projection.weight'),
              ('module.patch_embed.proj.bias', 'vit.embeddings.patch_embeddings.projection.bias'),
              ('module.pos_embed', 'vit.embeddings.position_embeddings'),
          ]							)

      if base_model:
            # layernorm + pooler
            rename_keys.extend(
                [
                    ('module.norm.weight', 'layernorm.weight'),
                    ('module.norm.bias', 'layernorm.bias'),
                ]							)

            # if just the base model, we should remove "vit" from all keys that start with "vit"
            SCREAMING_SNAKE_CASE_							=			[(pair[0], pair[1][4:]) if pair[1].startswith('vit'							) else pair for pair in rename_keys]
      else:
            # layernorm + classification head
            rename_keys.extend(
                [
                    ('norm.weight', 'vit.layernorm.weight'),
                    ('norm.bias', 'vit.layernorm.bias'),
                    ('head.weight', 'classifier.weight'),
                    ('head.bias', 'classifier.bias'),
                ]							)

      return rename_keys

def 					UpperCAmelCase_							(      __UpperCAmelCase						:	Union[str, Any]			,			__UpperCAmelCase						:	Union[str, Any]			,			__UpperCAmelCase						:	Optional[Any]=False							)					-> Any:
      for i in range(config.num_hidden_layers							):
            if base_model:
                  SCREAMING_SNAKE_CASE_							=			''
            else:
                  SCREAMING_SNAKE_CASE_							=			'vit.'
            # read in weights + bias of input projection layer (in timm, this is a single matrix + bias)
            SCREAMING_SNAKE_CASE_							=			state_dict.pop(f"module.blocks.{i}.attn.qkv.weight"							)
            SCREAMING_SNAKE_CASE_							=			state_dict.pop(f"module.blocks.{i}.attn.qkv.bias"							)
            # next, add query, keys and values (in that order) to the state dict
            SCREAMING_SNAKE_CASE_							=			in_proj_weight[
                : config.hidden_size, :
            ]
            SCREAMING_SNAKE_CASE_							=			in_proj_bias[: config.hidden_size]
            SCREAMING_SNAKE_CASE_							=			in_proj_weight[
                config.hidden_size : config.hidden_size * 2, :
            ]
            SCREAMING_SNAKE_CASE_							=			in_proj_bias[
                config.hidden_size : config.hidden_size * 2
            ]
            SCREAMING_SNAKE_CASE_							=			in_proj_weight[
                -config.hidden_size :, :
            ]
            SCREAMING_SNAKE_CASE_							=			in_proj_bias[-config.hidden_size :]

def 					UpperCAmelCase_							(      __UpperCAmelCase						:	List[Any]							)					-> Union[str, Any]:
      SCREAMING_SNAKE_CASE_							=			['head.weight', 'head.bias']
      for k in ignore_keys:
            state_dict.pop(__UpperCAmelCase			,			__UpperCAmelCase							)

def 					UpperCAmelCase_							(      __UpperCAmelCase						:	Optional[int]							)					-> str:
      # projection head is used in the self-supervised pre-training in MSN,
      # for downstream task it's not needed.
      SCREAMING_SNAKE_CASE_							=			[
          'module.fc.fc1.weight',
          'module.fc.fc1.bias',
          'module.fc.bn1.weight',
          'module.fc.bn1.bias',
          'module.fc.bn1.running_mean',
          'module.fc.bn1.running_var',
          'module.fc.bn1.num_batches_tracked',
          'module.fc.fc2.weight',
          'module.fc.fc2.bias',
          'module.fc.bn2.weight',
          'module.fc.bn2.bias',
          'module.fc.bn2.running_mean',
          'module.fc.bn2.running_var',
          'module.fc.bn2.num_batches_tracked',
          'module.fc.fc3.weight',
          'module.fc.fc3.bias',
      ]
      for k in ignore_keys:
            state_dict.pop(__UpperCAmelCase			,			__UpperCAmelCase							)

def 					UpperCAmelCase_							(      __UpperCAmelCase						:	Dict			,			__UpperCAmelCase						:	Any			,			__UpperCAmelCase						:	Any							)					-> Dict:
      SCREAMING_SNAKE_CASE_							=			dct.pop(__UpperCAmelCase							)
      SCREAMING_SNAKE_CASE_							=			val

def 					UpperCAmelCase_							(      __UpperCAmelCase						:	Optional[int]			,			__UpperCAmelCase						:	Tuple							)					-> List[str]:
      SCREAMING_SNAKE_CASE_							=			ViTMSNConfig()
      SCREAMING_SNAKE_CASE_							=			10_00

      SCREAMING_SNAKE_CASE_							=			'datasets/huggingface/label-files'
      SCREAMING_SNAKE_CASE_							=			'imagenet-1k-id2label.json'
      SCREAMING_SNAKE_CASE_							=			json.load(open(hf_hub_download(__UpperCAmelCase			,			__UpperCAmelCase							)			,			'r'							)							)
      SCREAMING_SNAKE_CASE_							=			{int(__UpperCAmelCase							): v for k, v in idalabel.items()}
      SCREAMING_SNAKE_CASE_							=			idalabel
      SCREAMING_SNAKE_CASE_							=			{v: k for k, v in idalabel.items()}

      if "s16" in checkpoint_url:
            SCREAMING_SNAKE_CASE_							=			3_84
            SCREAMING_SNAKE_CASE_							=			15_36
            SCREAMING_SNAKE_CASE_							=			6
      elif "l16" in checkpoint_url:
            SCREAMING_SNAKE_CASE_							=			10_24
            SCREAMING_SNAKE_CASE_							=			40_96
            SCREAMING_SNAKE_CASE_							=			24
            SCREAMING_SNAKE_CASE_							=			16
            SCREAMING_SNAKE_CASE_							=			0.1
      elif "b4" in checkpoint_url:
            SCREAMING_SNAKE_CASE_							=			4
      elif "l7" in checkpoint_url:
            SCREAMING_SNAKE_CASE_							=			7
            SCREAMING_SNAKE_CASE_							=			10_24
            SCREAMING_SNAKE_CASE_							=			40_96
            SCREAMING_SNAKE_CASE_							=			24
            SCREAMING_SNAKE_CASE_							=			16
            SCREAMING_SNAKE_CASE_							=			0.1

      SCREAMING_SNAKE_CASE_							=			ViTMSNModel(__UpperCAmelCase							)

      SCREAMING_SNAKE_CASE_							=			torch.hub.load_state_dict_from_url(__UpperCAmelCase			,			map_location='cpu'							)['target_encoder']

      SCREAMING_SNAKE_CASE_							=			ViTImageProcessor(size=config.image_size							)

      remove_projection_head(__UpperCAmelCase							)
      SCREAMING_SNAKE_CASE_							=			create_rename_keys(__UpperCAmelCase			,			base_model=__UpperCAmelCase							)

      for src, dest in rename_keys:
            rename_key(__UpperCAmelCase			,			__UpperCAmelCase			,			__UpperCAmelCase							)
      read_in_q_k_v(__UpperCAmelCase			,			__UpperCAmelCase			,			base_model=__UpperCAmelCase							)

      model.load_state_dict(__UpperCAmelCase							)
      model.eval()

      SCREAMING_SNAKE_CASE_							=			'http://images.cocodataset.org/val2017/000000039769.jpg'

      SCREAMING_SNAKE_CASE_							=			Image.open(requests.get(__UpperCAmelCase			,			stream=__UpperCAmelCase							).raw							)
      SCREAMING_SNAKE_CASE_							=			ViTImageProcessor(
          size=config.image_size			,			image_mean=__UpperCAmelCase			,			image_std=__UpperCAmelCase							)
      SCREAMING_SNAKE_CASE_							=			image_processor(images=__UpperCAmelCase			,			return_tensors='pt'							)

      # forward pass
      torch.manual_seed(2							)
      SCREAMING_SNAKE_CASE_							=			model(**__UpperCAmelCase							)
      SCREAMING_SNAKE_CASE_							=			outputs.last_hidden_state

      # The following Colab Notebook was used to generate these outputs:
      # https://colab.research.google.com/gist/sayakpaul/3672419a04f5997827503fd84079bdd1/scratchpad.ipynb
      if "s16" in checkpoint_url:
            SCREAMING_SNAKE_CASE_							=			torch.tensor([[-1.0_9_1_5, -1.4_8_7_6, -1.1_8_0_9]]							)
      elif "b16" in checkpoint_url:
            SCREAMING_SNAKE_CASE_							=			torch.tensor([[1_4.2_8_8_9, -1_8.9_0_4_5, 1_1.7_2_8_1]]							)
      elif "l16" in checkpoint_url:
            SCREAMING_SNAKE_CASE_							=			torch.tensor([[4_1.5_0_2_8, -2_2.8_6_8_1, 4_5.6_4_7_5]]							)
      elif "b4" in checkpoint_url:
            SCREAMING_SNAKE_CASE_							=			torch.tensor([[-4.3_8_6_8, 5.2_9_3_2, -0.4_1_3_7]]							)
      else:
            SCREAMING_SNAKE_CASE_							=			torch.tensor([[-0.1_7_9_2, -0.6_4_6_5, 2.4_2_6_3]]							)

      # verify logits
      assert torch.allclose(last_hidden_state[:, 0, :3]			,			__UpperCAmelCase			,			atol=1E-4							)

      print(f"Saving model to {pytorch_dump_folder_path}"							)
      model.save_pretrained(__UpperCAmelCase							)

      print(f"Saving image processor to {pytorch_dump_folder_path}"							)
      image_processor.save_pretrained(__UpperCAmelCase							)


if __name__ == "__main__":
   lowerCamelCase__ :   int					=							argparse.ArgumentParser()
   # Required parameters
   parser.add_argument(
       '--checkpoint_url',
       default='https://dl.fbaipublicfiles.com/msn/vits16_800ep.pth.tar',
       type=str,
       help='URL of the checkpoint you\'d like to convert.',
   )
   parser.add_argument(
       '--pytorch_dump_folder_path', default=None, type=str, help='Path to the output PyTorch model directory.'
   )

   lowerCamelCase__ :   Union[str, Any]					=							parser.parse_args()
   convert_vit_msn_checkpoint(args.checkpoint_url, args.pytorch_dump_folder_path)