

import argparse
import json

import gdown
import numpy as np
import torch
from huggingface_hub import hf_hub_download

from transformers import (
    VideoMAEConfig,
    VideoMAEForPreTraining,
    VideoMAEForVideoClassification,
    VideoMAEImageProcessor,
)






def 					__lowercase      (     _A			)     ->						Optional[Any]:
      SCREAMING_SNAKE_CASE :      Optional[int]       =   VideoMAEConfig()

      set_architecture_configs(_A						,    _A			)

      if "finetuned" not in model_name:
            SCREAMING_SNAKE_CASE :      Dict       =   False

      if "finetuned" in model_name:
            SCREAMING_SNAKE_CASE :      Any       =   """huggingface/label-files"""
            if "kinetics" in model_name:
                  SCREAMING_SNAKE_CASE :      Optional[int]       =   400
                  SCREAMING_SNAKE_CASE :      int       =   """kinetics400-id2label.json"""
            elif "ssv2" in model_name:
                  SCREAMING_SNAKE_CASE :      Any       =   174
                  SCREAMING_SNAKE_CASE :      List[Any]       =   """something-something-v2-id2label.json"""
            else:
                  raise ValueError("""Model name should either contain 'kinetics' or 'ssv2' in case it's fine-tuned."""			)
            SCREAMING_SNAKE_CASE :      int       =   json.load(open(hf_hub_download(_A						,    _A						,    repo_type="""dataset"""			)						,    """r"""			)			)
            SCREAMING_SNAKE_CASE :      List[Any]       =   {int(_A			): v for k, v in idalabel.items()}
            SCREAMING_SNAKE_CASE :      Tuple       =   idalabel
            SCREAMING_SNAKE_CASE :      int       =   {v: k for k, v in idalabel.items()}

      return config






def 					__lowercase      (     _A						,    _A			)     ->						Dict:
      if "small" in model_name:
            SCREAMING_SNAKE_CASE :      Dict       =   384
            SCREAMING_SNAKE_CASE :      Tuple       =   1536
            SCREAMING_SNAKE_CASE :      Optional[Any]       =   12
            SCREAMING_SNAKE_CASE :      Any       =   16
            SCREAMING_SNAKE_CASE :      Optional[Any]       =   12
            SCREAMING_SNAKE_CASE :      Tuple       =   3
            SCREAMING_SNAKE_CASE :      List[str]       =   192
            SCREAMING_SNAKE_CASE :      Tuple       =   768
      elif "large" in model_name:
            SCREAMING_SNAKE_CASE :      Dict       =   1024
            SCREAMING_SNAKE_CASE :      Optional[int]       =   4096
            SCREAMING_SNAKE_CASE :      List[Any]       =   24
            SCREAMING_SNAKE_CASE :      str       =   16
            SCREAMING_SNAKE_CASE :      Tuple       =   12
            SCREAMING_SNAKE_CASE :      int       =   8
            SCREAMING_SNAKE_CASE :      List[Any]       =   512
            SCREAMING_SNAKE_CASE :      int       =   2048
      elif "huge" in model_name:
            SCREAMING_SNAKE_CASE :      Tuple       =   1280
            SCREAMING_SNAKE_CASE :      int       =   5120
            SCREAMING_SNAKE_CASE :      Optional[Any]       =   32
            SCREAMING_SNAKE_CASE :      str       =   16
            SCREAMING_SNAKE_CASE :      List[Any]       =   12
            SCREAMING_SNAKE_CASE :      Tuple       =   8
            SCREAMING_SNAKE_CASE :      Tuple       =   640
            SCREAMING_SNAKE_CASE :      Optional[Any]       =   2560
      elif "base" not in model_name:
            raise ValueError("""Model name should include either \"small\", \"base\", \"large\", or \"huge\""""			)






def 					__lowercase      (     _A			)     ->						List[str]:
      if "encoder." in name:
            SCREAMING_SNAKE_CASE :      List[Any]       =   name.replace("""encoder."""						,    """"""			)
      if "cls_token" in name:
            SCREAMING_SNAKE_CASE :      Dict       =   name.replace("""cls_token"""						,    """videomae.embeddings.cls_token"""			)
      if "decoder_pos_embed" in name:
            SCREAMING_SNAKE_CASE :      List[str]       =   name.replace("""decoder_pos_embed"""						,    """decoder.decoder_pos_embed"""			)
      if "pos_embed" in name and "decoder" not in name:
            SCREAMING_SNAKE_CASE :      List[str]       =   name.replace("""pos_embed"""						,    """videomae.embeddings.position_embeddings"""			)
      if "patch_embed.proj" in name:
            SCREAMING_SNAKE_CASE :      Optional[Any]       =   name.replace("""patch_embed.proj"""						,    """videomae.embeddings.patch_embeddings.projection"""			)
      if "patch_embed.norm" in name:
            SCREAMING_SNAKE_CASE :      Optional[Any]       =   name.replace("""patch_embed.norm"""						,    """videomae.embeddings.norm"""			)
      if "decoder.blocks" in name:
            SCREAMING_SNAKE_CASE :      Optional[int]       =   name.replace("""decoder.blocks"""						,    """decoder.decoder_layers"""			)
      if "blocks" in name:
            SCREAMING_SNAKE_CASE :      int       =   name.replace("""blocks"""						,    """videomae.encoder.layer"""			)
      if "attn.proj" in name:
            SCREAMING_SNAKE_CASE :      Tuple       =   name.replace("""attn.proj"""						,    """attention.output.dense"""			)
      if "attn" in name and "bias" not in name:
            SCREAMING_SNAKE_CASE :      List[Any]       =   name.replace("""attn"""						,    """attention.self"""			)
      if "attn" in name:
            SCREAMING_SNAKE_CASE :      List[str]       =   name.replace("""attn"""						,    """attention.attention"""			)
      if "norm1" in name:
            SCREAMING_SNAKE_CASE :      Dict       =   name.replace("""norm1"""						,    """layernorm_before"""			)
      if "norm2" in name:
            SCREAMING_SNAKE_CASE :      Optional[Any]       =   name.replace("""norm2"""						,    """layernorm_after"""			)
      if "mlp.fc1" in name:
            SCREAMING_SNAKE_CASE :      Union[str, Any]       =   name.replace("""mlp.fc1"""						,    """intermediate.dense"""			)
      if "mlp.fc2" in name:
            SCREAMING_SNAKE_CASE :      str       =   name.replace("""mlp.fc2"""						,    """output.dense"""			)
      if "decoder_embed" in name:
            SCREAMING_SNAKE_CASE :      Dict       =   name.replace("""decoder_embed"""						,    """decoder.decoder_embed"""			)
      if "decoder_norm" in name:
            SCREAMING_SNAKE_CASE :      Tuple       =   name.replace("""decoder_norm"""						,    """decoder.decoder_norm"""			)
      if "decoder_pred" in name:
            SCREAMING_SNAKE_CASE :      str       =   name.replace("""decoder_pred"""						,    """decoder.decoder_pred"""			)
      if "norm.weight" in name and "decoder" not in name and "fc" not in name:
            SCREAMING_SNAKE_CASE :      List[Any]       =   name.replace("""norm.weight"""						,    """videomae.layernorm.weight"""			)
      if "norm.bias" in name and "decoder" not in name and "fc" not in name:
            SCREAMING_SNAKE_CASE :      List[str]       =   name.replace("""norm.bias"""						,    """videomae.layernorm.bias"""			)
      if "head" in name and "decoder" not in name:
            SCREAMING_SNAKE_CASE :      Optional[int]       =   name.replace("""head"""						,    """classifier"""			)

      return name






def 					__lowercase      (     _A						,    _A			)     ->						str:
      for key in orig_state_dict.copy().keys():
            SCREAMING_SNAKE_CASE :      Optional[Any]       =   orig_state_dict.pop(_A			)

            if key.startswith("""encoder."""			):
                  SCREAMING_SNAKE_CASE :      Dict       =   key.replace("""encoder."""						,    """"""			)

            if "qkv" in key:
                  SCREAMING_SNAKE_CASE :      Union[str, Any]       =   key.split("""."""			)
                  if key.startswith("""decoder.blocks"""			):
                        SCREAMING_SNAKE_CASE :      Union[str, Any]       =   config.decoder_hidden_size
                        SCREAMING_SNAKE_CASE :      Dict       =   int(key_split[2]			)
                        SCREAMING_SNAKE_CASE :      Optional[Any]       =   """decoder.decoder_layers."""
                        if "weight" in key:
                              SCREAMING_SNAKE_CASE :      Tuple       =   val[:dim, :]
                              SCREAMING_SNAKE_CASE :      Tuple       =   val[dim : dim * 2, :]
                              SCREAMING_SNAKE_CASE :      Optional[Any]       =   val[-dim:, :]
                  else:
                        SCREAMING_SNAKE_CASE :      List[str]       =   config.hidden_size
                        SCREAMING_SNAKE_CASE :      Dict       =   int(key_split[1]			)
                        SCREAMING_SNAKE_CASE :      str       =   """videomae.encoder.layer."""
                        if "weight" in key:
                              SCREAMING_SNAKE_CASE :      List[Any]       =   val[:dim, :]
                              SCREAMING_SNAKE_CASE :      Tuple       =   val[dim : dim * 2, :]
                              SCREAMING_SNAKE_CASE :      int       =   val[-dim:, :]
            else:
                  SCREAMING_SNAKE_CASE :      Union[str, Any]       =   val

      return orig_state_dict






def 					__lowercase      (     )     ->						Union[str, Any]:
      SCREAMING_SNAKE_CASE :      List[str]       =   hf_hub_download(
          repo_id="""hf-internal-testing/spaghetti-video"""						,    filename="""eating_spaghetti.npy"""						,    repo_type="""dataset"""			)
      SCREAMING_SNAKE_CASE :      Dict       =   np.load(_A			)
      return list(_A			)






def 					__lowercase      (     _A						,    _A						,    _A						,    _A			)     ->						List[str]:
      SCREAMING_SNAKE_CASE :      Union[str, Any]       =   get_videomae_config(_A			)

      if "finetuned" in model_name:
            SCREAMING_SNAKE_CASE :      str       =   VideoMAEForVideoClassification(_A			)
      else:
            SCREAMING_SNAKE_CASE :      Any       =   VideoMAEForPreTraining(_A			)

      # download original checkpoint, hosted on Google Drive
      SCREAMING_SNAKE_CASE :      Optional[Any]       =   """pytorch_model.bin"""
      gdown.cached_download(_A						,    _A						,    quiet=_A			)
      SCREAMING_SNAKE_CASE :      Optional[int]       =   torch.load(_A						,    map_location="""cpu"""			)
      if "model" in files:
            SCREAMING_SNAKE_CASE :      int       =   files["""model"""]
      else:
            SCREAMING_SNAKE_CASE :      str       =   files["""module"""]
      SCREAMING_SNAKE_CASE :      List[Any]       =   convert_state_dict(_A						,    _A			)

      model.load_state_dict(_A			)
      model.eval()

      # verify model on basic input
      SCREAMING_SNAKE_CASE :      Optional[Any]       =   VideoMAEImageProcessor(image_mean=[0.5, 0.5, 0.5]						,    image_std=[0.5, 0.5, 0.5]			)
      SCREAMING_SNAKE_CASE :      Dict       =   prepare_video()
      SCREAMING_SNAKE_CASE :      List[str]       =   image_processor(_A						,    return_tensors="""pt"""			)

      if "finetuned" not in model_name:
            SCREAMING_SNAKE_CASE :      int       =   hf_hub_download(repo_id="""hf-internal-testing/bool-masked-pos"""						,    filename="""bool_masked_pos.pt"""			)
            SCREAMING_SNAKE_CASE :      List[str]       =   torch.load(_A			)

      SCREAMING_SNAKE_CASE :      str       =   model(**_A			)
      SCREAMING_SNAKE_CASE :      str       =   outputs.logits

      SCREAMING_SNAKE_CASE :      int       =   [
          """videomae-small-finetuned-kinetics""",
          """videomae-small-finetuned-ssv2""",
          # Kinetics-400 checkpoints (short = pretrained only for 800 epochs instead of 1600)
          """videomae-base-short""",
          """videomae-base-short-finetuned-kinetics""",
          """videomae-base""",
          """videomae-base-finetuned-kinetics""",
          """videomae-large""",
          """videomae-large-finetuned-kinetics""",
          """videomae-huge-finetuned-kinetics""",
          # Something-Something-v2 checkpoints (short = pretrained only for 800 epochs instead of 2400)
          """videomae-base-short-ssv2""",
          """videomae-base-short-finetuned-ssv2""",
          """videomae-base-ssv2""",
          """videomae-base-finetuned-ssv2""",
      ]

      # NOTE: logits were tested with image_mean and image_std equal to [0.5, 0.5, 0.5] and [0.5, 0.5, 0.5]
      if model_name == "videomae-small-finetuned-kinetics":
            SCREAMING_SNAKE_CASE :      Union[str, Any]       =   torch.Size([1, 400]			)
            SCREAMING_SNAKE_CASE :      List[Any]       =   torch.tensor([-0.9291, -0.4061, -0.9307]			)
      elif model_name == "videomae-small-finetuned-ssv2":
            SCREAMING_SNAKE_CASE :      List[Any]       =   torch.Size([1, 174]			)
            SCREAMING_SNAKE_CASE :      Union[str, Any]       =   torch.tensor([0.2671, -0.4689, -0.8235]			)
      elif model_name == "videomae-base":
            SCREAMING_SNAKE_CASE :      List[str]       =   torch.Size([1, 1408, 1536]			)
            SCREAMING_SNAKE_CASE :      Dict       =   torch.tensor([[0.7739, 0.7968, 0.7089], [0.6701, 0.7487, 0.6209], [0.4287, 0.5158, 0.4773]]			)
      elif model_name == "videomae-base-short":
            SCREAMING_SNAKE_CASE :      Dict       =   torch.Size([1, 1408, 1536]			)
            SCREAMING_SNAKE_CASE :      List[Any]       =   torch.tensor([[0.7994, 0.9612, 0.8508], [0.7401, 0.8958, 0.8302], [0.5862, 0.7468, 0.7325]]			)
            # we verified the loss both for normalized and unnormalized targets for this one
            SCREAMING_SNAKE_CASE :      List[str]       =   torch.tensor([0.5142]			) if config.norm_pix_loss else torch.tensor([0.6469]			)
      elif model_name == "videomae-large":
            SCREAMING_SNAKE_CASE :      str       =   torch.Size([1, 1408, 1536]			)
            SCREAMING_SNAKE_CASE :      Optional[int]       =   torch.tensor([[0.7149, 0.7997, 0.6966], [0.6768, 0.7869, 0.6948], [0.5139, 0.6221, 0.5605]]			)
      elif model_name == "videomae-large-finetuned-kinetics":
            SCREAMING_SNAKE_CASE :      Any       =   torch.Size([1, 400]			)
            SCREAMING_SNAKE_CASE :      Optional[int]       =   torch.tensor([0.0771, 0.0011, -0.3625]			)
      elif model_name == "videomae-huge-finetuned-kinetics":
            SCREAMING_SNAKE_CASE :      Dict       =   torch.Size([1, 400]			)
            SCREAMING_SNAKE_CASE :      str       =   torch.tensor([0.2433, 0.1632, -0.4894]			)
      elif model_name == "videomae-base-short-finetuned-kinetics":
            SCREAMING_SNAKE_CASE :      Tuple       =   torch.Size([1, 400]			)
            SCREAMING_SNAKE_CASE :      Optional[Any]       =   torch.tensor([0.6588, 0.0990, -0.2493]			)
      elif model_name == "videomae-base-finetuned-kinetics":
            SCREAMING_SNAKE_CASE :      Optional[Any]       =   torch.Size([1, 400]			)
            SCREAMING_SNAKE_CASE :      List[str]       =   torch.tensor([0.3669, -0.0688, -0.2421]			)
      elif model_name == "videomae-base-short-ssv2":
            SCREAMING_SNAKE_CASE :      Any       =   torch.Size([1, 1408, 1536]			)
            SCREAMING_SNAKE_CASE :      Optional[Any]       =   torch.tensor([[0.4712, 0.5296, 0.5786], [0.2278, 0.2729, 0.4026], [0.0352, 0.0730, 0.2506]]			)
      elif model_name == "videomae-base-short-finetuned-ssv2":
            SCREAMING_SNAKE_CASE :      Any       =   torch.Size([1, 174]			)
            SCREAMING_SNAKE_CASE :      Optional[Any]       =   torch.tensor([-0.0537, -0.1539, -0.3266]			)
      elif model_name == "videomae-base-ssv2":
            SCREAMING_SNAKE_CASE :      Any       =   torch.Size([1, 1408, 1536]			)
            SCREAMING_SNAKE_CASE :      Optional[int]       =   torch.tensor([[0.8131, 0.8727, 0.8546], [0.7366, 0.9377, 0.8870], [0.5935, 0.8874, 0.8564]]			)
      elif model_name == "videomae-base-finetuned-ssv2":
            SCREAMING_SNAKE_CASE :      Dict       =   torch.Size([1, 174]			)
            SCREAMING_SNAKE_CASE :      Union[str, Any]       =   torch.tensor([0.1961, -0.8337, -0.6389]			)
      else:
            raise ValueError(F"Model name not supported. Should be one of {model_names}"			)

      # verify logits
      assert logits.shape == expected_shape
      if "finetuned" in model_name:
            assert torch.allclose(logits[0, :3]						,    _A						,    atol=1e-4			)
      else:
            print("""Logits:"""						,    logits[0, :3, :3]			)
            assert torch.allclose(logits[0, :3, :3]						,    _A						,    atol=1e-4			)
      print("""Logits ok!"""			)

      # verify loss, if applicable
      if model_name == "videomae-base-short":
            SCREAMING_SNAKE_CASE :      Optional[Any]       =   outputs.loss
            assert torch.allclose(_A						,    _A						,    atol=1e-4			)
            print("""Loss ok!"""			)

      if pytorch_dump_folder_path is not None:
            print(F"Saving model and image processor to {pytorch_dump_folder_path}"			)
            image_processor.save_pretrained(_A			)
            model.save_pretrained(_A			)

      if push_to_hub:
            print("""Pushing to the hub..."""			)
            model.push_to_hub(_A						,    organization="""nielsr"""			)


if __name__ == "__main__":
  UpperCAmelCase__ :     Tuple				    =    argparse.ArgumentParser()
  # Required parameters
  parser.add_argument(
      """--checkpoint_url""",
      default="""https://drive.google.com/u/1/uc?id=1tEhLyskjb755TJ65ptsrafUG2llSwQE1&amp;export=download&amp;confirm=t&amp;uuid=aa3276eb-fb7e-482a-adec-dc7171df14c4""",
      type=str,
      help=(
          """URL of the original PyTorch checkpoint (on Google Drive) you'd like to convert. Should be a direct"""
          """ download link."""
      ),
  )
  parser.add_argument(
      """--pytorch_dump_folder_path""",
      default="""/Users/nielsrogge/Documents/VideoMAE/Test""",
      type=str,
      help="""Path to the output PyTorch model directory.""",
  )
  parser.add_argument("""--model_name""", default="""videomae-base""", type=str, help="""Name of the model.""")
  parser.add_argument(
      """--push_to_hub""", action="""store_true""", help="""Whether or not to push the converted model to the 🤗 hub."""
  )

  UpperCAmelCase__ :     List[str]				    =    parser.parse_args()
  convert_videomae_checkpoint(args.checkpoint_url, args.pytorch_dump_folder_path, args.model_name, args.push_to_hub)







import inspect
from typing import Callable, List, Optional, Union

import torch
from transformers import CLIPImageProcessor, CLIPTextModel, CLIPTokenizer

from diffusers import DiffusionPipeline
from diffusers.models import AutoencoderKL, UNetaDConditionModel
from diffusers.pipelines.stable_diffusion import StableDiffusionPipelineOutput
from diffusers.pipelines.stable_diffusion.safety_checker import StableDiffusionSafetyChecker
from diffusers.schedulers import DDIMScheduler, LMSDiscreteScheduler, PNDMScheduler
from diffusers.utils import logging


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







class        a__       (   UpperCAmelCase       ):







  """simple docstring"""




  def __init__(			self    : Dict				,				UpperCAmelCase__    : AutoencoderKL				,				UpperCAmelCase__    : CLIPTextModel				,				UpperCAmelCase__    : CLIPTokenizer				,				UpperCAmelCase__    : UNetaDConditionModel				,				UpperCAmelCase__    : Union[DDIMScheduler, PNDMScheduler, LMSDiscreteScheduler]				,				UpperCAmelCase__    : StableDiffusionSafetyChecker				,				UpperCAmelCase__    : CLIPImageProcessor				,				)					->Optional[Any]:
        """simple docstring"""







        super().__init__()
        self.register_modules(
            vae=UpperCAmelCase__				,				text_encoder=UpperCAmelCase__				,				tokenizer=UpperCAmelCase__				,				unet=UpperCAmelCase__				,				scheduler=UpperCAmelCase__				,				safety_checker=UpperCAmelCase__				,				feature_extractor=UpperCAmelCase__				,				)


  def  _lowercase      (			self    : int				,				UpperCAmelCase__    : Optional[Union[str, int]] = "auto"		)					->Any:
        """simple docstring"""







        if slice_size == "auto":
              # half the attention head size is usually a good trade-off between
              # speed and memory
              SCREAMING_SNAKE_CASE :      int       =   self.unet.config.attention_head_dim // 2
        self.unet.set_attention_slice(UpperCAmelCase__		)


  def  _lowercase      (			self    : Dict		)					->Tuple:
        """simple docstring"""







        self.enable_attention_slicing(UpperCAmelCase__		)




  @torch.no_grad()
  def __call__(			self    : Optional[Any]				,				UpperCAmelCase__    : Union[str, List[str]]				,				UpperCAmelCase__    : int = 5_1_2				,				UpperCAmelCase__    : int = 5_1_2				,				UpperCAmelCase__    : int = 5_0				,				UpperCAmelCase__    : float = 7.5				,				UpperCAmelCase__    : Optional[Union[str, List[str]]] = None				,				UpperCAmelCase__    : Optional[int] = 1				,				UpperCAmelCase__    : float = 0.0				,				UpperCAmelCase__    : Optional[torch.Generator] = None				,				UpperCAmelCase__    : Optional[torch.FloatTensor] = None				,				UpperCAmelCase__    : Optional[str] = "pil"				,				UpperCAmelCase__    : bool = True				,				UpperCAmelCase__    : Optional[Callable[[int, int, torch.FloatTensor], None]] = None				,				UpperCAmelCase__    : int = 1				,				UpperCAmelCase__    : Optional[torch.FloatTensor] = None				,				**UpperCAmelCase__    : Tuple				,				)					->List[Any]:
        """simple docstring"""







        if isinstance(UpperCAmelCase__				,				UpperCAmelCase__		):
              SCREAMING_SNAKE_CASE :      Any       =   1
        elif isinstance(UpperCAmelCase__				,				UpperCAmelCase__		):
              SCREAMING_SNAKE_CASE :      Any       =   len(UpperCAmelCase__		)
        else:
              raise ValueError(f"`prompt` has to be of type `str` or `list` but is {type(UpperCAmelCase__		)}"		)

        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 (callback_steps is None) or (
            callback_steps is not None and (not isinstance(UpperCAmelCase__				,				UpperCAmelCase__		) or callback_steps <= 0)
        ):
              raise ValueError(
                  f"`callback_steps` has to be a positive integer but is {callback_steps} of type"
                  f" {type(UpperCAmelCase__		)}."		)

        # get prompt text embeddings
        SCREAMING_SNAKE_CASE :      Optional[int]       =   self.tokenizer(
            UpperCAmelCase__				,				padding="""max_length"""				,				max_length=self.tokenizer.model_max_length				,				return_tensors="""pt"""				,				)
        SCREAMING_SNAKE_CASE :      str       =   text_inputs.input_ids

        if text_input_ids.shape[-1] > self.tokenizer.model_max_length:
              SCREAMING_SNAKE_CASE :      Union[str, Any]       =   self.tokenizer.batch_decode(text_input_ids[:, self.tokenizer.model_max_length :]		)
              logger.warning(
                  """The following part of your input was truncated because CLIP can only handle sequences up to"""
                  f" {self.tokenizer.model_max_length} tokens: {removed_text}"		)
              SCREAMING_SNAKE_CASE :      Tuple       =   text_input_ids[:, : self.tokenizer.model_max_length]

        if text_embeddings is None:
              SCREAMING_SNAKE_CASE :      Any       =   self.text_encoder(text_input_ids.to(self.device		)		)[0]

        # duplicate text embeddings for each generation per prompt, using mps friendly method
        SCREAMING_SNAKE_CASE     ,	SCREAMING_SNAKE_CASE     ,	SCREAMING_SNAKE_CASE :      str       =   text_embeddings.shape
        SCREAMING_SNAKE_CASE :      Tuple       =   text_embeddings.repeat(1				,				UpperCAmelCase__				,				1		)
        SCREAMING_SNAKE_CASE :      int       =   text_embeddings.view(bs_embed * num_images_per_prompt				,				UpperCAmelCase__				,				-1		)

        # 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 :      List[str]       =   guidance_scale > 1.0
        # get unconditional embeddings for classifier free guidance
        if do_classifier_free_guidance:
              SCREAMING_SNAKE_CASE :      List[str]
              if negative_prompt is None:
                    SCREAMING_SNAKE_CASE :      Union[str, Any]       =   [""""""]
              elif type(UpperCAmelCase__		) is not type(UpperCAmelCase__		):
                    raise TypeError(
                        f"`negative_prompt` should be the same type to `prompt`, but got {type(UpperCAmelCase__		)} !="
                        f" {type(UpperCAmelCase__		)}."		)
              elif isinstance(UpperCAmelCase__				,				UpperCAmelCase__		):
                    SCREAMING_SNAKE_CASE :      Union[str, Any]       =   [negative_prompt]
              elif batch_size != len(UpperCAmelCase__		):
                    raise ValueError(
                        f"`negative_prompt`: {negative_prompt} has batch size {len(UpperCAmelCase__		)}, but `prompt`:"
                        f" {prompt} has batch size {batch_size}. Please make sure that passed `negative_prompt` matches"
                        """ the batch size of `prompt`."""		)
              else:
                    SCREAMING_SNAKE_CASE :      Optional[int]       =   negative_prompt

              SCREAMING_SNAKE_CASE :      Dict       =   text_input_ids.shape[-1]
              SCREAMING_SNAKE_CASE :      int       =   self.tokenizer(
                  UpperCAmelCase__				,				padding="""max_length"""				,				max_length=UpperCAmelCase__				,				truncation=UpperCAmelCase__				,				return_tensors="""pt"""				,				)
              SCREAMING_SNAKE_CASE :      Union[str, Any]       =   self.text_encoder(uncond_input.input_ids.to(self.device		)		)[0]

              # duplicate unconditional embeddings for each generation per prompt, using mps friendly method
              SCREAMING_SNAKE_CASE :      Optional[int]       =   uncond_embeddings.shape[1]
              SCREAMING_SNAKE_CASE :      List[str]       =   uncond_embeddings.repeat(UpperCAmelCase__				,				UpperCAmelCase__				,				1		)
              SCREAMING_SNAKE_CASE :      int       =   uncond_embeddings.view(batch_size * num_images_per_prompt				,				UpperCAmelCase__				,				-1		)

              # 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 :      int       =   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 :      Dict       =   (batch_size * num_images_per_prompt, self.unet.config.in_channels, height // 8, width // 8)
        SCREAMING_SNAKE_CASE :      Dict       =   (batch_size * num_images_per_prompt, self.unet.config.in_channels, 6_4, 6_4)
        SCREAMING_SNAKE_CASE :      List[str]       =   text_embeddings.dtype
        if latents is None:
              if self.device.type == "mps":
                    # randn does not exist on mps
                    SCREAMING_SNAKE_CASE :      Union[str, Any]       =   torch.randn(
                        UpperCAmelCase__				,				generator=UpperCAmelCase__				,				device="""cpu"""				,				dtype=UpperCAmelCase__		).to(self.device		)
                    SCREAMING_SNAKE_CASE :      Dict       =   torch.randn(UpperCAmelCase__				,				generator=UpperCAmelCase__				,				device="""cpu"""				,				dtype=UpperCAmelCase__		).to(
                        self.device		)
              else:
                    SCREAMING_SNAKE_CASE :      List[Any]       =   torch.randn(
                        UpperCAmelCase__				,				generator=UpperCAmelCase__				,				device=self.device				,				dtype=UpperCAmelCase__		)
                    SCREAMING_SNAKE_CASE :      Any       =   torch.randn(UpperCAmelCase__				,				generator=UpperCAmelCase__				,				device=self.device				,				dtype=UpperCAmelCase__		)
        else:
              if latents_reference.shape != latents_shape:
                    raise ValueError(f"Unexpected latents shape, got {latents.shape}, expected {latents_shape}"		)
              SCREAMING_SNAKE_CASE :      List[str]       =   latents_reference.to(self.device		)
              SCREAMING_SNAKE_CASE :      Any       =   latents.to(self.device		)

        # This is the key part of the pipeline where we
        # try to ensure that the generated images w/ the same seed
        # but different sizes actually result in similar images
        SCREAMING_SNAKE_CASE :      Optional[Any]       =   (latents_shape[3] - latents_shape_reference[3]) // 2
        SCREAMING_SNAKE_CASE :      Tuple       =   (latents_shape[2] - latents_shape_reference[2]) // 2
        SCREAMING_SNAKE_CASE :      str       =   latents_shape_reference[3] if dx >= 0 else latents_shape_reference[3] + 2 * dx
        SCREAMING_SNAKE_CASE :      List[Any]       =   latents_shape_reference[2] if dy >= 0 else latents_shape_reference[2] + 2 * dy
        SCREAMING_SNAKE_CASE :      Optional[Any]       =   0 if dx < 0 else dx
        SCREAMING_SNAKE_CASE :      Any       =   0 if dy < 0 else dy
        SCREAMING_SNAKE_CASE :      int       =   max(-dx				,				0		)
        SCREAMING_SNAKE_CASE :      Optional[Any]       =   max(-dy				,				0		)
        # import pdb
        # pdb.set_trace()
        SCREAMING_SNAKE_CASE :      Tuple       =   latents_reference[:, :, dy : dy + h, dx : dx + w]

        # set timesteps
        self.scheduler.set_timesteps(UpperCAmelCase__		)

        # Some schedulers like PNDM have timesteps as arrays
        # It's more optimized to move all timesteps to correct device beforehand
        SCREAMING_SNAKE_CASE :      str       =   self.scheduler.timesteps.to(self.device		)

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

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

              # predict the noise residual
              SCREAMING_SNAKE_CASE :      Optional[int]       =   self.unet(UpperCAmelCase__				,				UpperCAmelCase__				,				encoder_hidden_states=UpperCAmelCase__		).sample

              # perform guidance
              if do_classifier_free_guidance:
                    SCREAMING_SNAKE_CASE     ,	SCREAMING_SNAKE_CASE :      Union[str, Any]       =   noise_pred.chunk(2		)
                    SCREAMING_SNAKE_CASE :      Optional[Any]       =   noise_pred_uncond + guidance_scale * (noise_pred_text - noise_pred_uncond)

              # compute the previous noisy sample x_t -> x_t-1
              SCREAMING_SNAKE_CASE :      Dict       =   self.scheduler.step(UpperCAmelCase__				,				UpperCAmelCase__				,				UpperCAmelCase__				,				**UpperCAmelCase__		).prev_sample

              # call the callback, if provided
              if callback is not None and i % callback_steps == 0:
                    callback(UpperCAmelCase__				,				UpperCAmelCase__				,				UpperCAmelCase__		)

        SCREAMING_SNAKE_CASE :      Union[str, Any]       =   1 / 0.1_82_15 * latents
        SCREAMING_SNAKE_CASE :      Tuple       =   self.vae.decode(UpperCAmelCase__		).sample

        SCREAMING_SNAKE_CASE :      Tuple       =   (image / 2 + 0.5).clamp(0				,				1		)

        # we always cast to float32 as this does not cause significant overhead and is compatible with bfloat16
        SCREAMING_SNAKE_CASE :      Dict       =   image.cpu().permute(0				,				2				,				3				,				1		).float().numpy()

        if self.safety_checker is not None:
              SCREAMING_SNAKE_CASE :      Tuple       =   self.feature_extractor(self.numpy_to_pil(UpperCAmelCase__		)				,				return_tensors="""pt"""		).to(
                  self.device		)
              SCREAMING_SNAKE_CASE     ,	SCREAMING_SNAKE_CASE :      Any       =   self.safety_checker(
                  images=UpperCAmelCase__				,				clip_input=safety_checker_input.pixel_values.to(text_embeddings.dtype		)		)
        else:
              SCREAMING_SNAKE_CASE :      Optional[Any]       =   None

        if output_type == "pil":
              SCREAMING_SNAKE_CASE :      Any       =   self.numpy_to_pil(UpperCAmelCase__		)

        if not return_dict:
              return (image, has_nsfw_concept)

        return StableDiffusionPipelineOutput(images=UpperCAmelCase__				,				nsfw_content_detected=UpperCAmelCase__		)




