

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 					__lowercase      (     _A						,    _A						,    _A			)     ->						int:
      if isinstance(_A						,    torch.Tensor			):
            return image
      elif isinstance(_A						,    PIL.Image.Image			):
            SCREAMING_SNAKE_CASE :      Any       =   [image]

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






def 					__lowercase      (     _A						,    _A						,    _A						,    _A=0.9995			)     ->						List[str]:
      if not isinstance(_A						,    np.ndarray			):
            SCREAMING_SNAKE_CASE :      int       =   True
            SCREAMING_SNAKE_CASE :      Optional[Any]       =   va.device
            SCREAMING_SNAKE_CASE :      Tuple       =   va.cpu().numpy()
            SCREAMING_SNAKE_CASE :      Dict       =   va.cpu().numpy()

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

      if inputs_are_torch:
            SCREAMING_SNAKE_CASE :      int       =   torch.from_numpy(_A			).to(_A			)

      return va






def 					__lowercase      (     _A						,    _A			)     ->						Union[str, Any]:
      SCREAMING_SNAKE_CASE :      Any       =   F.normalize(_A						,    dim=-1			)
      SCREAMING_SNAKE_CASE :      Union[str, Any]       =   F.normalize(_A						,    dim=-1			)
      return (x - y).norm(dim=-1			).div(2			).arcsin().pow(2			).mul(2			)






def 					__lowercase      (     _A						,    _A			)     ->						int:
      for param in model.parameters():
            SCREAMING_SNAKE_CASE :      Any       =   value







class        a__       (   UpperCAmelCase       ):







  """simple docstring"""




  def __init__(			self    : str				,				UpperCAmelCase__    : AutoencoderKL				,				UpperCAmelCase__    : CLIPTextModel				,				UpperCAmelCase__    : CLIPModel				,				UpperCAmelCase__    : CLIPTokenizer				,				UpperCAmelCase__    : UNetaDConditionModel				,				UpperCAmelCase__    : Union[PNDMScheduler, LMSDiscreteScheduler, DDIMScheduler, DPMSolverMultistepScheduler]				,				UpperCAmelCase__    : CLIPFeatureExtractor				,				UpperCAmelCase__    : Optional[int]=None				,				UpperCAmelCase__    : Dict=None				,				UpperCAmelCase__    : Optional[Any]=None				,				)					->Dict:
        """simple docstring"""







        super().__init__()
        self.register_modules(
            vae=UpperCAmelCase__				,				text_encoder=UpperCAmelCase__				,				clip_model=UpperCAmelCase__				,				tokenizer=UpperCAmelCase__				,				unet=UpperCAmelCase__				,				scheduler=UpperCAmelCase__				,				feature_extractor=UpperCAmelCase__				,				coca_model=UpperCAmelCase__				,				coca_tokenizer=UpperCAmelCase__				,				coca_transform=UpperCAmelCase__				,				)
        SCREAMING_SNAKE_CASE :      Dict       =   (
            feature_extractor.size
            if isinstance(feature_extractor.size				,				UpperCAmelCase__		)
            else feature_extractor.size["""shortest_edge"""]
        )
        SCREAMING_SNAKE_CASE :      int       =   transforms.Normalize(mean=feature_extractor.image_mean				,				std=feature_extractor.image_std		)
        set_requires_grad(self.text_encoder				,				UpperCAmelCase__		)
        set_requires_grad(self.clip_model				,				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 :      List[str]       =   self.unet.config.attention_head_dim // 2
        self.unet.set_attention_slice(UpperCAmelCase__		)


  def  _lowercase      (			self    : str		)					->int:
        """simple docstring"""







        self.enable_attention_slicing(UpperCAmelCase__		)


  def  _lowercase      (			self    : Dict		)					->Optional[int]:
        """simple docstring"""







        set_requires_grad(self.vae				,				UpperCAmelCase__		)


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







        set_requires_grad(self.vae				,				UpperCAmelCase__		)


  def  _lowercase      (			self    : List[str]		)					->List[Any]:
        """simple docstring"""







        set_requires_grad(self.unet				,				UpperCAmelCase__		)


  def  _lowercase      (			self    : Optional[int]		)					->Tuple:
        """simple docstring"""







        set_requires_grad(self.unet				,				UpperCAmelCase__		)


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







        SCREAMING_SNAKE_CASE :      int       =   min(int(num_inference_steps * strength		)				,				UpperCAmelCase__		)

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

        return timesteps, num_inference_steps - t_start


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







        if not isinstance(UpperCAmelCase__				,				torch.Tensor		):
              raise ValueError(f"`image` has to be of type `torch.Tensor` but is {type(UpperCAmelCase__		)}"		)

        SCREAMING_SNAKE_CASE :      Optional[Any]       =   image.to(device=UpperCAmelCase__				,				dtype=UpperCAmelCase__		)

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

        # Hardcode 0.18215 because stable-diffusion-2-base has not self.vae.config.scaling_factor
        SCREAMING_SNAKE_CASE :      str       =   0.1_82_15 * init_latents
        SCREAMING_SNAKE_CASE :      int       =   init_latents.repeat_interleave(UpperCAmelCase__				,				dim=0		)

        SCREAMING_SNAKE_CASE :      List[Any]       =   randn_tensor(init_latents.shape				,				generator=UpperCAmelCase__				,				device=UpperCAmelCase__				,				dtype=UpperCAmelCase__		)

        # get latents
        SCREAMING_SNAKE_CASE :      str       =   self.scheduler.add_noise(UpperCAmelCase__				,				UpperCAmelCase__				,				UpperCAmelCase__		)
        SCREAMING_SNAKE_CASE :      List[Any]       =   init_latents

        return latents


  def  _lowercase      (			self    : int				,				UpperCAmelCase__    : List[Any]		)					->Tuple:
        """simple docstring"""







        SCREAMING_SNAKE_CASE :      Tuple       =   self.coca_transform(UpperCAmelCase__		).unsqueeze(0		)
        with torch.no_grad(), torch.cuda.amp.autocast():
              SCREAMING_SNAKE_CASE :      List[str]       =   self.coca_model.generate(transformed_image.to(device=self.device				,				dtype=self.coca_model.dtype		)		)
        SCREAMING_SNAKE_CASE :      Optional[int]       =   self.coca_tokenizer.decode(generated[0].cpu().numpy()		)
        return generated.split("""<end_of_text>"""		)[0].replace("""<start_of_text>"""				,				""""""		).rstrip(""" .,"""		)


  def  _lowercase      (			self    : Any				,				UpperCAmelCase__    : Any				,				UpperCAmelCase__    : str		)					->Any:
        """simple docstring"""







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


  @torch.enable_grad()
  def  _lowercase      (			self    : Dict				,				UpperCAmelCase__    : str				,				UpperCAmelCase__    : Any				,				UpperCAmelCase__    : str				,				UpperCAmelCase__    : List[Any]				,				UpperCAmelCase__    : Union[str, Any]				,				UpperCAmelCase__    : Tuple				,				UpperCAmelCase__    : str				,				)					->str:
        """simple docstring"""







        SCREAMING_SNAKE_CASE :      Optional[int]       =   latents.detach().requires_grad_()

        SCREAMING_SNAKE_CASE :      List[Any]       =   self.scheduler.scale_model_input(UpperCAmelCase__				,				UpperCAmelCase__		)

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

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

              SCREAMING_SNAKE_CASE :      Any       =   torch.sqrt(UpperCAmelCase__		)
              SCREAMING_SNAKE_CASE :      List[Any]       =   pred_original_sample * (fac) + latents * (1 - fac)
        elif isinstance(self.scheduler				,				UpperCAmelCase__		):
              SCREAMING_SNAKE_CASE :      int       =   self.scheduler.sigmas[index]
              SCREAMING_SNAKE_CASE :      Optional[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
        SCREAMING_SNAKE_CASE :      List[str]       =   1 / 0.1_82_15 * sample
        SCREAMING_SNAKE_CASE :      Optional[Any]       =   self.vae.decode(UpperCAmelCase__		).sample
        SCREAMING_SNAKE_CASE :      str       =   (image / 2 + 0.5).clamp(0				,				1		)

        SCREAMING_SNAKE_CASE :      str       =   transforms.Resize(self.feature_extractor_size		)(UpperCAmelCase__		)
        SCREAMING_SNAKE_CASE :      List[str]       =   self.normalize(UpperCAmelCase__		).to(latents.dtype		)

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

        SCREAMING_SNAKE_CASE :      str       =   spherical_dist_loss(UpperCAmelCase__				,				UpperCAmelCase__		).mean() * clip_guidance_scale

        SCREAMING_SNAKE_CASE :      List[str]       =   -torch.autograd.grad(UpperCAmelCase__				,				UpperCAmelCase__		)[0]

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




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







        if isinstance(UpperCAmelCase__				,				UpperCAmelCase__		) and len(UpperCAmelCase__		) != batch_size:
              raise ValueError(f"You have passed {batch_size} batch_size, but only {len(UpperCAmelCase__		)} 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(UpperCAmelCase__				,				torch.Generator		) and batch_size > 1:
              SCREAMING_SNAKE_CASE :      List[Any]       =   [generator] + [None] * (batch_size - 1)

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

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

        SCREAMING_SNAKE_CASE :      str       =   self.tokenizer(
            UpperCAmelCase__				,				padding="""max_length"""				,				max_length=self.tokenizer.model_max_length				,				truncation=UpperCAmelCase__				,				return_tensors="""pt"""				,				)
        SCREAMING_SNAKE_CASE :      Optional[Any]       =   self.text_encoder(style_text_input.input_ids.to(self.device		)		)[0]

        SCREAMING_SNAKE_CASE :      Optional[Any]       =   slerp(UpperCAmelCase__				,				UpperCAmelCase__				,				UpperCAmelCase__		)

        # duplicate text embeddings for each generation per prompt
        SCREAMING_SNAKE_CASE :      Tuple       =   text_embeddings.repeat_interleave(UpperCAmelCase__				,				dim=0		)

        # set timesteps
        SCREAMING_SNAKE_CASE :      Tuple       =   """offset""" in set(inspect.signature(self.scheduler.set_timesteps		).parameters.keys()		)
        SCREAMING_SNAKE_CASE :      Optional[Any]       =   {}
        if accepts_offset:
              SCREAMING_SNAKE_CASE :      int       =   1

        self.scheduler.set_timesteps(UpperCAmelCase__				,				**UpperCAmelCase__		)
        # 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 :      str       =   self.get_timesteps(UpperCAmelCase__				,				UpperCAmelCase__				,				self.device		)
        SCREAMING_SNAKE_CASE :      List[str]       =   timesteps[:1].repeat(UpperCAmelCase__		)

        # Preprocess image
        SCREAMING_SNAKE_CASE :      Any       =   preprocess(UpperCAmelCase__				,				UpperCAmelCase__				,				UpperCAmelCase__		)
        SCREAMING_SNAKE_CASE :      Any       =   self.prepare_latents(
            UpperCAmelCase__				,				UpperCAmelCase__				,				UpperCAmelCase__				,				text_embeddings.dtype				,				self.device				,				UpperCAmelCase__		)

        SCREAMING_SNAKE_CASE :      Any       =   preprocess(UpperCAmelCase__				,				UpperCAmelCase__				,				UpperCAmelCase__		)
        SCREAMING_SNAKE_CASE :      str       =   self.prepare_latents(
            UpperCAmelCase__				,				UpperCAmelCase__				,				UpperCAmelCase__				,				text_embeddings.dtype				,				self.device				,				UpperCAmelCase__		)

        SCREAMING_SNAKE_CASE :      List[Any]       =   slerp(UpperCAmelCase__				,				UpperCAmelCase__				,				UpperCAmelCase__		)

        if clip_guidance_scale > 0:
              SCREAMING_SNAKE_CASE :      List[Any]       =   self.get_clip_image_embeddings(UpperCAmelCase__				,				UpperCAmelCase__		)
              SCREAMING_SNAKE_CASE :      List[str]       =   self.get_clip_image_embeddings(UpperCAmelCase__				,				UpperCAmelCase__		)
              SCREAMING_SNAKE_CASE :      int       =   slerp(
                  UpperCAmelCase__				,				UpperCAmelCase__				,				UpperCAmelCase__		)

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

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

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

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

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

                    # perform classifier free guidance
                    if do_classifier_free_guidance:
                          SCREAMING_SNAKE_CASE     ,	SCREAMING_SNAKE_CASE :      Dict       =   noise_pred.chunk(2		)
                          SCREAMING_SNAKE_CASE :      List[str]       =   noise_pred_uncond + guidance_scale * (noise_pred_text - noise_pred_uncond)

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

                    # compute the previous noisy sample x_t -> x_t-1
                    SCREAMING_SNAKE_CASE :      Union[str, Any]       =   self.scheduler.step(UpperCAmelCase__				,				UpperCAmelCase__				,				UpperCAmelCase__				,				**UpperCAmelCase__		).prev_sample

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

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

        if output_type == "pil":
              SCREAMING_SNAKE_CASE :      Optional[int]       =   self.numpy_to_pil(UpperCAmelCase__		)

        if not return_dict:
              return (image, None)

        return StableDiffusionPipelineOutput(images=UpperCAmelCase__				,				nsfw_content_detected=UpperCAmelCase__		)







def 					__lowercase      (     _A						,    _A			)     ->						int:
      return int((input_a, input_a).count(1			) != 0			)






def 					__lowercase      (     )     ->						None:
      assert or_gate(0						,    0			) == 0
      assert or_gate(0						,    1			) == 1
      assert or_gate(1						,    0			) == 1
      assert or_gate(1						,    1			) == 1


if __name__ == "__main__":
  print(or_gate(0, 1))
  print(or_gate(1, 0))
  print(or_gate(0, 0))
  print(or_gate(1, 1))




