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 			A__  (	snake_case_ :		int       ,				snake_case_ :		List[str]       ,				snake_case_ :		int      ):
 if isinstance(snake_case_       ,				torch.Tensor      ):
  return image
 elif isinstance(snake_case_       ,				PIL.Image.Image      ):
  SCREAMING_SNAKE_CASE__:   Union[str, Any]=      [image]

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







def 			A__  (	snake_case_ :		Dict       ,				snake_case_ :		Union[str, Any]       ,				snake_case_ :		Union[str, Any]       ,				snake_case_ :		Any=0.99_95      ):
 if not isinstance(snake_case_       ,				np.ndarray      ):
  SCREAMING_SNAKE_CASE__:   List[str]=      True
  SCREAMING_SNAKE_CASE__:   Any=      va.device
  SCREAMING_SNAKE_CASE__:   str=      va.cpu().numpy()
  SCREAMING_SNAKE_CASE__:   Union[str, Any]=      va.cpu().numpy()

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

 if inputs_are_torch:
  SCREAMING_SNAKE_CASE__:   Optional[Any]=      torch.from_numpy(snake_case_      ).to(snake_case_      )

 return va







def 			A__  (	snake_case_ :		Union[str, Any]       ,				snake_case_ :		Optional[Any]      ):
 SCREAMING_SNAKE_CASE__:   Optional[Any]=      F.normalize(snake_case_       ,				dim=-1      )
 SCREAMING_SNAKE_CASE__:   List[Any]=      F.normalize(snake_case_       ,				dim=-1      )
 return (x - y).norm(dim=-1      ).div(2      ).arcsin().pow(2      ).mul(2      )







def 			A__  (	snake_case_ :		Dict       ,				snake_case_ :		List[Any]      ):
 for param in model.parameters():
  SCREAMING_SNAKE_CASE__:   Any=      value






class 		_lowerCamelCase		(   UpperCamelCase_    ):
    def __init__(  self , lowerCAmelCase , lowerCAmelCase , lowerCAmelCase , lowerCAmelCase , lowerCAmelCase , lowerCAmelCase , lowerCAmelCase , lowerCAmelCase=None , lowerCAmelCase=None , lowerCAmelCase=None , )       ->					str:
     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__:   Optional[Any]=      (
         feature_extractor.size
         if isinstance(feature_extractor.size , lowerCAmelCase   )
         else feature_extractor.size['''shortest_edge''']
     )
     SCREAMING_SNAKE_CASE__:   Any=      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 						UpperCamelCase_				(  self , lowerCAmelCase = "auto"   )       ->					Dict:
     if slice_size == "auto":
      # half the attention head size is usually a good trade-off between
      # speed and memory
      SCREAMING_SNAKE_CASE__:   Optional[int]=      self.unet.config.attention_head_dim // 2
     self.unet.set_attention_slice(lowerCAmelCase   )
    def 						UpperCamelCase_				(  self   )       ->					int:
     self.enable_attention_slicing(lowerCAmelCase   )
    def 						UpperCamelCase_				(  self   )       ->					str:
     set_requires_grad(self.vae , lowerCAmelCase   )
    def 						UpperCamelCase_				(  self   )       ->					Any:
     set_requires_grad(self.vae , lowerCAmelCase   )
    def 						UpperCamelCase_				(  self   )       ->					Optional[Any]:
     set_requires_grad(self.unet , lowerCAmelCase   )
    def 						UpperCamelCase_				(  self   )       ->					Union[str, Any]:
     set_requires_grad(self.unet , lowerCAmelCase   )
    def 						UpperCamelCase_				(  self , lowerCAmelCase , lowerCAmelCase , lowerCAmelCase   )       ->					Optional[int]:
     # get the original timestep using init_timestep
     SCREAMING_SNAKE_CASE__:   int=      min(int(num_inference_steps * strength   ) , lowerCAmelCase   )

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

     return timesteps, num_inference_steps - t_start
    def 						UpperCamelCase_				(  self , lowerCAmelCase , lowerCAmelCase , lowerCAmelCase , lowerCAmelCase , lowerCAmelCase , lowerCAmelCase=None   )       ->					Any:
     if not isinstance(lowerCAmelCase , torch.Tensor   ):
      raise ValueError(f'`image` has to be of type `torch.Tensor` but is {type(lowerCAmelCase   )}'   )

     SCREAMING_SNAKE_CASE__:   int=      image.to(device=lowerCAmelCase , dtype=lowerCAmelCase   )

     if isinstance(lowerCAmelCase , lowerCAmelCase   ):
      SCREAMING_SNAKE_CASE__:   Optional[int]=      [
          self.vae.encode(image[i : i + 1]   ).latent_dist.sample(generator[i]   ) for i in range(lowerCAmelCase   )
      ]
      SCREAMING_SNAKE_CASE__:   List[str]=      torch.cat(lowerCAmelCase , dim=0   )
     else:
      SCREAMING_SNAKE_CASE__:   List[str]=      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__:   str=      0.18215 * init_latents
     SCREAMING_SNAKE_CASE__:   Union[str, Any]=      init_latents.repeat_interleave(lowerCAmelCase , dim=0   )

     SCREAMING_SNAKE_CASE__:   List[str]=      randn_tensor(init_latents.shape , generator=lowerCAmelCase , device=lowerCAmelCase , dtype=lowerCAmelCase   )

     # get latents
     SCREAMING_SNAKE_CASE__:   Tuple=      self.scheduler.add_noise(lowerCAmelCase , lowerCAmelCase , lowerCAmelCase   )
     SCREAMING_SNAKE_CASE__:   List[Any]=      init_latents

     return latents
    def 						UpperCamelCase_				(  self , lowerCAmelCase   )       ->					Any:
     SCREAMING_SNAKE_CASE__:   Union[str, Any]=      self.coca_transform(lowerCAmelCase   ).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__:   str=      self.coca_tokenizer.decode(generated[0].cpu().numpy()   )
     return generated.split('''<end_of_text>'''   )[0].replace('''<start_of_text>''' , ''''''   ).rstrip(''' .,'''   )
    def 						UpperCamelCase_				(  self , lowerCAmelCase , lowerCAmelCase   )       ->					List[str]:
     SCREAMING_SNAKE_CASE__:   str=      self.feature_extractor.preprocess(lowerCAmelCase   )
     SCREAMING_SNAKE_CASE__:   List[str]=      torch.from_numpy(clip_image_input['''pixel_values'''][0]   ).unsqueeze(0   ).to(self.device   ).half()
     SCREAMING_SNAKE_CASE__:   Optional[Any]=      self.clip_model.get_image_features(lowerCAmelCase   )
     SCREAMING_SNAKE_CASE__:   Optional[int]=      image_embeddings_clip / image_embeddings_clip.norm(p=2 , dim=-1 , keepdim=lowerCAmelCase   )
     SCREAMING_SNAKE_CASE__:   str=      image_embeddings_clip.repeat_interleave(lowerCAmelCase , dim=0   )
     return image_embeddings_clip
    @torch.enable_grad()
    def 						UpperCamelCase_				(  self , lowerCAmelCase , lowerCAmelCase , lowerCAmelCase , lowerCAmelCase , lowerCAmelCase , lowerCAmelCase , lowerCAmelCase , )       ->					Tuple:
     SCREAMING_SNAKE_CASE__:   List[Any]=      latents.detach().requires_grad_()

     SCREAMING_SNAKE_CASE__:   Union[str, Any]=      self.scheduler.scale_model_input(lowerCAmelCase , lowerCAmelCase   )

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

     if isinstance(self.scheduler , (PNDMScheduler, DDIMScheduler, DPMSolverMultistepScheduler)   ):
      SCREAMING_SNAKE_CASE__:   Dict=      self.scheduler.alphas_cumprod[timestep]
      SCREAMING_SNAKE_CASE__:   Optional[Any]=      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__:   Optional[Any]=      torch.sqrt(lowerCAmelCase   )
      SCREAMING_SNAKE_CASE__:   int=      pred_original_sample * (fac) + latents * (1 - fac)
     elif isinstance(self.scheduler , lowerCAmelCase   ):
      SCREAMING_SNAKE_CASE__:   Dict=      self.scheduler.sigmas[index]
      SCREAMING_SNAKE_CASE__:   Optional[int]=      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[Any]=      1 / 0.18215 * sample
     SCREAMING_SNAKE_CASE__:   Tuple=      self.vae.decode(lowerCAmelCase   ).sample
     SCREAMING_SNAKE_CASE__:   Optional[int]=      (image / 2 + 0.5).clamp(0 , 1   )

     SCREAMING_SNAKE_CASE__:   int=      transforms.Resize(self.feature_extractor_size   )(lowerCAmelCase   )
     SCREAMING_SNAKE_CASE__:   Tuple=      self.normalize(lowerCAmelCase   ).to(latents.dtype   )

     SCREAMING_SNAKE_CASE__:   int=      self.clip_model.get_image_features(lowerCAmelCase   )
     SCREAMING_SNAKE_CASE__:   Dict=      image_embeddings_clip / image_embeddings_clip.norm(p=2 , dim=-1 , keepdim=lowerCAmelCase   )

     SCREAMING_SNAKE_CASE__:   Union[str, Any]=      spherical_dist_loss(lowerCAmelCase , lowerCAmelCase   ).mean() * clip_guidance_scale

     SCREAMING_SNAKE_CASE__:   Tuple=      -torch.autograd.grad(lowerCAmelCase , lowerCAmelCase   )[0]

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



    @torch.no_grad()
    def __call__(  self , lowerCAmelCase , lowerCAmelCase , lowerCAmelCase = None , lowerCAmelCase = None , lowerCAmelCase = 512 , lowerCAmelCase = 512 , lowerCAmelCase = 0.6 , lowerCAmelCase = 50 , lowerCAmelCase = 7.5 , lowerCAmelCase = 1 , lowerCAmelCase = 0.0 , lowerCAmelCase = 100 , lowerCAmelCase = None , lowerCAmelCase = "pil" , lowerCAmelCase = True , lowerCAmelCase = 0.8 , lowerCAmelCase = 0.1 , lowerCAmelCase = 0.1 , )       ->					Optional[Any]:
     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__:   Optional[Any]=      [generator] + [None] * (batch_size - 1)

     SCREAMING_SNAKE_CASE__:   Optional[Any]=      [
         ('''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__:   Dict=      ''', '''.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__:   List[Any]=      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__:   Optional[Any]=      self.get_image_description(lowerCAmelCase   )

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

     SCREAMING_SNAKE_CASE__:   str=      self.tokenizer(
         lowerCAmelCase , padding='''max_length''' , max_length=self.tokenizer.model_max_length , truncation=lowerCAmelCase , return_tensors='''pt''' , )
     SCREAMING_SNAKE_CASE__:   Optional[int]=      self.text_encoder(style_text_input.input_ids.to(self.device   )   )[0]

     SCREAMING_SNAKE_CASE__:   List[Any]=      slerp(lowerCAmelCase , lowerCAmelCase , lowerCAmelCase   )

     # duplicate text embeddings for each generation per prompt
     SCREAMING_SNAKE_CASE__:   Optional[Any]=      text_embeddings.repeat_interleave(lowerCAmelCase , dim=0   )

     # set timesteps
     SCREAMING_SNAKE_CASE__:   Optional[Any]=      '''offset''' in set(inspect.signature(self.scheduler.set_timesteps   ).parameters.keys()   )
     SCREAMING_SNAKE_CASE__:   Tuple=      {}
     if accepts_offset:
      SCREAMING_SNAKE_CASE__:   Union[str, Any]=      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__:   Optional[Any]=      self.get_timesteps(lowerCAmelCase , lowerCAmelCase , self.device   )
     SCREAMING_SNAKE_CASE__:   Tuple=      timesteps[:1].repeat(lowerCAmelCase   )

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

     SCREAMING_SNAKE_CASE__:   Dict=      preprocess(lowerCAmelCase , lowerCAmelCase , lowerCAmelCase   )
     SCREAMING_SNAKE_CASE__:   int=      self.prepare_latents(
         lowerCAmelCase , lowerCAmelCase , lowerCAmelCase , text_embeddings.dtype , self.device , lowerCAmelCase   )

     SCREAMING_SNAKE_CASE__:   str=      slerp(lowerCAmelCase , lowerCAmelCase , lowerCAmelCase   )

     if clip_guidance_scale > 0:
      SCREAMING_SNAKE_CASE__:   Optional[Any]=      self.get_clip_image_embeddings(lowerCAmelCase , lowerCAmelCase   )
      SCREAMING_SNAKE_CASE__:   Any=      self.get_clip_image_embeddings(lowerCAmelCase , lowerCAmelCase   )
      SCREAMING_SNAKE_CASE__:   Dict=      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__:   Any=      guidance_scale > 1.0
     # get unconditional embeddings for classifier free guidance
     if do_classifier_free_guidance:
      SCREAMING_SNAKE_CASE__:   Optional[Any]=      content_text_input.input_ids.shape[-1]
      SCREAMING_SNAKE_CASE__:   Union[str, Any]=      self.tokenizer([''''''] , padding='''max_length''' , max_length=lowerCAmelCase , return_tensors='''pt'''   )
      SCREAMING_SNAKE_CASE__:   Optional[Any]=      self.text_encoder(uncond_input.input_ids.to(self.device   )   )[0]
      # duplicate unconditional embeddings for each generation per prompt
      SCREAMING_SNAKE_CASE__:   int=      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__:   List[str]=      torch.cat([uncond_embeddings, text_embeddings]   )

     # get the initial random noise unless the user supplied it

     # Unlike in other pipelines, latents need to be generated in the target device
     # for 1-to-1 results reproducibility with the CompVis implementation.
     # However this currently doesn't work in `mps`.
     SCREAMING_SNAKE_CASE__:   Optional[int]=      (batch_size, self.unet.config.in_channels, height // 8, width // 8)
     SCREAMING_SNAKE_CASE__:   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(lowerCAmelCase , generator=lowerCAmelCase , device='''cpu''' , dtype=lowerCAmelCase   ).to(
           self.device   )
      else:
       SCREAMING_SNAKE_CASE__:   str=      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__:   str=      latents.to(self.device   )

     # scale the initial noise by the standard deviation required by the scheduler
     SCREAMING_SNAKE_CASE__:   Any=      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__:   Optional[Any]=      {}
     if accepts_eta:
      SCREAMING_SNAKE_CASE__:   List[Any]=      eta

     # check if the scheduler accepts generator
     SCREAMING_SNAKE_CASE__:   Optional[Any]=      '''generator''' in set(inspect.signature(self.scheduler.step   ).parameters.keys()   )
     if accepts_generator:
      SCREAMING_SNAKE_CASE__:   Tuple=      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__:   str=      torch.cat([latents] * 2   ) if do_classifier_free_guidance else latents
       SCREAMING_SNAKE_CASE__:   List[str]=      self.scheduler.scale_model_input(lowerCAmelCase , lowerCAmelCase   )

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

       # perform classifier free guidance
       if do_classifier_free_guidance:
        SCREAMING_SNAKE_CASE__:   Any=      noise_pred.chunk(2   )
        SCREAMING_SNAKE_CASE__:   Optional[int]=      noise_pred_uncond + guidance_scale * (noise_pred_text - noise_pred_uncond)

       # perform clip guidance
       if clip_guidance_scale > 0:
        SCREAMING_SNAKE_CASE__:   Union[str, Any]=      (
            text_embeddings.chunk(2   )[1] if do_classifier_free_guidance else text_embeddings
        )
        SCREAMING_SNAKE_CASE__:   int=      self.cond_fn(
            lowerCAmelCase , lowerCAmelCase , lowerCAmelCase , lowerCAmelCase , lowerCAmelCase , lowerCAmelCase , lowerCAmelCase , )

       # compute the previous noisy sample x_t -> x_t-1
       SCREAMING_SNAKE_CASE__:   int=      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__:   List[Any]=      1 / 0.18215 * latents
     SCREAMING_SNAKE_CASE__:   Tuple=      self.vae.decode(lowerCAmelCase   ).sample

     SCREAMING_SNAKE_CASE__:   Tuple=      (image / 2 + 0.5).clamp(0 , 1   )
     SCREAMING_SNAKE_CASE__:   int=      image.cpu().permute(0 , 2 , 3 , 1   ).numpy()

     if output_type == "pil":
      SCREAMING_SNAKE_CASE__:   Tuple=      self.numpy_to_pil(lowerCAmelCase   )

     if not return_dict:
      return (image, None)

     return StableDiffusionPipelineOutput(images=lowerCAmelCase , nsfw_content_detected=lowerCAmelCase   )


# tests directory-specific settings - this file is run automatically
# by pytest before any tests are run

import sys
import warnings
from os.path import abspath, dirname, join


# allow having multiple repository checkouts and not needing to remember to rerun
# 'pip install -e .[dev]' when switching between checkouts and running tests.
lowercase_     :   List[str] 				=				abspath(join(dirname(dirname(__file__)), 'src'))
sys.path.insert(1, git_repo_path)

# silence FutureWarning warnings in tests since often we can't act on them until
# they become normal warnings - i.e. the tests still need to test the current functionality
warnings.simplefilter(action='ignore', category=FutureWarning)







def 			A__      (	snake_case_ :		Optional[int]      ):
 from diffusers.utils.testing_utils import pytest_addoption_shared

 pytest_addoption_shared(snake_case_      )







def 			A__      (	snake_case_ :		Optional[int]      ):
 from diffusers.utils.testing_utils import pytest_terminal_summary_main

 SCREAMING_SNAKE_CASE__:   Optional[int]=      terminalreporter.config.getoption('''--make-reports'''      )
 if make_reports:
  pytest_terminal_summary_main(snake_case_       ,				id=snake_case_      )

