

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_     (							a    ,      a    ,      a     ):

      """simple docstring"""



      if isinstance(a    ,      torch.Tensor     ):
            return image
      elif isinstance(a    ,      PIL.Image.Image     ):
            SCREAMING_SNAKE_CASE_ :     List[str]							=   [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_ :     Any							=   np.concatenate(a    ,      axis=0     )
            SCREAMING_SNAKE_CASE_ :     Optional[Any]							=   np.array(a     ).astype(np.floataa     ) / 2_5_5.0
            SCREAMING_SNAKE_CASE_ :     Dict							=   image.transpose(0    ,      3    ,      1    ,      2     )
            SCREAMING_SNAKE_CASE_ :     Union[str, Any]							=   2.0 * image - 1.0
            SCREAMING_SNAKE_CASE_ :     List[str]							=   torch.from_numpy(a     )
      elif isinstance(image[0]    ,      torch.Tensor     ):
            SCREAMING_SNAKE_CASE_ :     Dict							=   torch.cat(a    ,      dim=0     )
      return image

def       A_     (							a    ,      a    ,      a    ,      a=0.99_95     ):

      """simple docstring"""



      if not isinstance(a    ,      np.ndarray     ):
            SCREAMING_SNAKE_CASE_ :     Union[str, Any]							=   True
            SCREAMING_SNAKE_CASE_ :     List[str]							=   va.device
            SCREAMING_SNAKE_CASE_ :     List[str]							=   va.cpu().numpy()
            SCREAMING_SNAKE_CASE_ :     Dict							=   va.cpu().numpy()

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

      if inputs_are_torch:
            SCREAMING_SNAKE_CASE_ :     List[Any]							=   torch.from_numpy(a     ).to(a     )

      return va

def       A_     (							a    ,      a     ):

      """simple docstring"""



      SCREAMING_SNAKE_CASE_ :     int							=   F.normalize(a    ,      dim=-1     )
      SCREAMING_SNAKE_CASE_ :     Dict							=   F.normalize(a    ,      dim=-1     )
      return (x - y).norm(dim=-1     ).div(2     ).arcsin().pow(2     ).mul(2     )

def       A_     (							a    ,      a     ):

      """simple docstring"""



      for param in model.parameters():
            SCREAMING_SNAKE_CASE_ :     Dict							=   value



class        _A						(     a__):

  def __init__( self       ,    _SCREAMING_SNAKE_CASE       ,    _SCREAMING_SNAKE_CASE       ,    _SCREAMING_SNAKE_CASE       ,    _SCREAMING_SNAKE_CASE       ,    _SCREAMING_SNAKE_CASE       ,    _SCREAMING_SNAKE_CASE       ,    _SCREAMING_SNAKE_CASE       ,    _SCREAMING_SNAKE_CASE=None       ,    _SCREAMING_SNAKE_CASE=None       ,    _SCREAMING_SNAKE_CASE=None       ,    ):





        """simple docstring"""







        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_ :     int							=   (
            feature_extractor.size
            if isinstance(feature_extractor.size       ,    __lowerCAmelCase    )
            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       ,    __lowerCAmelCase    )
        set_requires_grad(self.clip_model       ,    __lowerCAmelCase    )

  def        UpperCAmelCase			( self       ,    _SCREAMING_SNAKE_CASE = "auto"    ):





        """simple docstring"""







        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    ):





        """simple docstring"""







        self.enable_attention_slicing(__lowerCAmelCase    )

  def        UpperCAmelCase			( self    ):





        """simple docstring"""







        set_requires_grad(self.vae       ,    __lowerCAmelCase    )

  def        UpperCAmelCase			( self    ):





        """simple docstring"""







        set_requires_grad(self.vae       ,    __lowerCAmelCase    )

  def        UpperCAmelCase			( self    ):





        """simple docstring"""







        set_requires_grad(self.unet       ,    __lowerCAmelCase    )

  def        UpperCAmelCase			( self    ):





        """simple docstring"""







        set_requires_grad(self.unet       ,    __lowerCAmelCase    )

  def        UpperCAmelCase			( self       ,    _SCREAMING_SNAKE_CASE       ,    _SCREAMING_SNAKE_CASE       ,    _SCREAMING_SNAKE_CASE    ):





        """simple docstring"""







        SCREAMING_SNAKE_CASE_ :     List[Any]							=   min(int(num_inference_steps * strength    )       ,    __lowerCAmelCase    )

        SCREAMING_SNAKE_CASE_ :     Tuple							=   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        UpperCAmelCase			( self       ,    _SCREAMING_SNAKE_CASE       ,    _SCREAMING_SNAKE_CASE       ,    _SCREAMING_SNAKE_CASE       ,    _SCREAMING_SNAKE_CASE       ,    _SCREAMING_SNAKE_CASE       ,    _SCREAMING_SNAKE_CASE=None    ):





        """simple docstring"""







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

        SCREAMING_SNAKE_CASE_ :     Optional[int]							=   image.to(device=__lowerCAmelCase       ,    dtype=__lowerCAmelCase    )

        if isinstance(__lowerCAmelCase       ,    __lowerCAmelCase    ):
              SCREAMING_SNAKE_CASE_ :     List[str]							=   [
                  self.vae.encode(image[i : i + 1]    ).latent_dist.sample(generator[i]    ) for i in range(__lowerCAmelCase    )
              ]
              SCREAMING_SNAKE_CASE_ :     List[Any]							=   torch.cat(__lowerCAmelCase       ,    dim=0    )
        else:
              SCREAMING_SNAKE_CASE_ :     Tuple							=   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_ :     Dict							=   0.18215 * init_latents
        SCREAMING_SNAKE_CASE_ :     List[str]							=   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_ :     Any							=   self.scheduler.add_noise(__lowerCAmelCase       ,    __lowerCAmelCase       ,    __lowerCAmelCase    )
        SCREAMING_SNAKE_CASE_ :     Union[str, Any]							=   init_latents

        return latents

  def        UpperCAmelCase			( self       ,    _SCREAMING_SNAKE_CASE    ):





        """simple docstring"""







        SCREAMING_SNAKE_CASE_ :     Optional[Any]							=   self.coca_transform(__lowerCAmelCase    ).unsqueeze(0    )
        with torch.no_grad(), torch.cuda.amp.autocast():
              SCREAMING_SNAKE_CASE_ :     Any							=   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       ,    _SCREAMING_SNAKE_CASE       ,    _SCREAMING_SNAKE_CASE    ):





        """simple docstring"""







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

  @torch.enable_grad()
  def        UpperCAmelCase			( self       ,    _SCREAMING_SNAKE_CASE       ,    _SCREAMING_SNAKE_CASE       ,    _SCREAMING_SNAKE_CASE       ,    _SCREAMING_SNAKE_CASE       ,    _SCREAMING_SNAKE_CASE       ,    _SCREAMING_SNAKE_CASE       ,    _SCREAMING_SNAKE_CASE       ,    ):





        """simple docstring"""







        SCREAMING_SNAKE_CASE_ :     int							=   latents.detach().requires_grad_()

        SCREAMING_SNAKE_CASE_ :     int							=   self.scheduler.scale_model_input(__lowerCAmelCase       ,    __lowerCAmelCase    )

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

        if isinstance(self.scheduler       ,    (PNDMScheduler, DDIMScheduler, DPMSolverMultistepScheduler)    ):
              SCREAMING_SNAKE_CASE_ :     Union[str, Any]							=   self.scheduler.alphas_cumprod[timestep]
              SCREAMING_SNAKE_CASE_ :     Dict							=   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_ :     str							=   pred_original_sample * (fac) + latents * (1 - fac)
        elif isinstance(self.scheduler       ,    __lowerCAmelCase    ):
              SCREAMING_SNAKE_CASE_ :     List[Any]							=   self.scheduler.sigmas[index]
              SCREAMING_SNAKE_CASE_ :     Union[str, 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_ :     Optional[Any]							=   1 / 0.18215 * sample
        SCREAMING_SNAKE_CASE_ :     List[str]							=   self.vae.decode(__lowerCAmelCase    ).sample
        SCREAMING_SNAKE_CASE_ :     Dict							=   (image / 2 + 0.5).clamp(0       ,    1    )

        SCREAMING_SNAKE_CASE_ :     Optional[Any]							=   transforms.Resize(self.feature_extractor_size    )(__lowerCAmelCase    )
        SCREAMING_SNAKE_CASE_ :     Optional[Any]							=   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_ :     str							=   -torch.autograd.grad(__lowerCAmelCase       ,    __lowerCAmelCase    )[0]

        if isinstance(self.scheduler       ,    __lowerCAmelCase    ):
              SCREAMING_SNAKE_CASE_ :     Dict							=   latents.detach() + grads * (sigma**2)
              SCREAMING_SNAKE_CASE_ :     Dict							=   noise_pred_original
        else:
              SCREAMING_SNAKE_CASE_ :     Optional[Any]							=   noise_pred_original - torch.sqrt(__lowerCAmelCase    ) * grads
        return noise_pred, latents



  @torch.no_grad()
  def __call__( self       ,    _SCREAMING_SNAKE_CASE       ,    _SCREAMING_SNAKE_CASE       ,    _SCREAMING_SNAKE_CASE = None       ,    _SCREAMING_SNAKE_CASE = None       ,    _SCREAMING_SNAKE_CASE = 512       ,    _SCREAMING_SNAKE_CASE = 512       ,    _SCREAMING_SNAKE_CASE = 0.6       ,    _SCREAMING_SNAKE_CASE = 50       ,    _SCREAMING_SNAKE_CASE = 7.5       ,    _SCREAMING_SNAKE_CASE = 1       ,    _SCREAMING_SNAKE_CASE = 0.0       ,    _SCREAMING_SNAKE_CASE = 100       ,    _SCREAMING_SNAKE_CASE = None       ,    _SCREAMING_SNAKE_CASE = "pil"       ,    _SCREAMING_SNAKE_CASE = True       ,    _SCREAMING_SNAKE_CASE = 0.8       ,    _SCREAMING_SNAKE_CASE = 0.1       ,    _SCREAMING_SNAKE_CASE = 0.1       ,    ):





        """simple docstring"""







        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_ :     List[Any]							=   [generator] + [None] * (batch_size - 1)

        SCREAMING_SNAKE_CASE_ :     Any							=   [
            ('''model''', self.coca_model is None),
            ('''tokenizer''', self.coca_tokenizer is None),
            ('''transform''', self.coca_transform is None),
        ]
        SCREAMING_SNAKE_CASE_ :     Dict							=   [x[0] for x in coca_is_none if x[1]]
        SCREAMING_SNAKE_CASE_ :     Optional[Any]							=   ''', '''.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_ :     Optional[int]							=   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[int]							=   self.get_image_description(__lowerCAmelCase    )

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

        SCREAMING_SNAKE_CASE_ :     int							=   self.tokenizer(
            __lowerCAmelCase       ,    padding='max_length'       ,    max_length=self.tokenizer.model_max_length       ,    truncation=__lowerCAmelCase       ,    return_tensors='pt'       ,    )
        SCREAMING_SNAKE_CASE_ :     Union[str, Any]							=   self.text_encoder(style_text_input.input_ids.to(self.device    )    )[0]

        SCREAMING_SNAKE_CASE_ :     Dict							=   slerp(__lowerCAmelCase       ,    __lowerCAmelCase       ,    __lowerCAmelCase    )

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

        # set timesteps
        SCREAMING_SNAKE_CASE_ :     List[Any]							=   '''offset''' in set(inspect.signature(self.scheduler.set_timesteps    ).parameters.keys()    )
        SCREAMING_SNAKE_CASE_ :     List[Any]							=   {}
        if accepts_offset:
              SCREAMING_SNAKE_CASE_ :     str							=   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_ :     int							=   self.get_timesteps(__lowerCAmelCase       ,    __lowerCAmelCase       ,    self.device    )
        SCREAMING_SNAKE_CASE_ :     int							=   timesteps[:1].repeat(__lowerCAmelCase    )

        # Preprocess image
        SCREAMING_SNAKE_CASE_ :     Union[str, Any]							=   preprocess(__lowerCAmelCase       ,    __lowerCAmelCase       ,    __lowerCAmelCase    )
        SCREAMING_SNAKE_CASE_ :     List[str]							=   self.prepare_latents(
            __lowerCAmelCase       ,    __lowerCAmelCase       ,    __lowerCAmelCase       ,    text_embeddings.dtype       ,    self.device       ,    __lowerCAmelCase    )

        SCREAMING_SNAKE_CASE_ :     Optional[int]							=   preprocess(__lowerCAmelCase       ,    __lowerCAmelCase       ,    __lowerCAmelCase    )
        SCREAMING_SNAKE_CASE_ :     Dict							=   self.prepare_latents(
            __lowerCAmelCase       ,    __lowerCAmelCase       ,    __lowerCAmelCase       ,    text_embeddings.dtype       ,    self.device       ,    __lowerCAmelCase    )

        SCREAMING_SNAKE_CASE_ :     Tuple							=   slerp(__lowerCAmelCase       ,    __lowerCAmelCase       ,    __lowerCAmelCase    )

        if clip_guidance_scale > 0:
              SCREAMING_SNAKE_CASE_ :     List[Any]							=   self.get_clip_image_embeddings(__lowerCAmelCase       ,    __lowerCAmelCase    )
              SCREAMING_SNAKE_CASE_ :     List[Any]							=   self.get_clip_image_embeddings(__lowerCAmelCase       ,    __lowerCAmelCase    )
              SCREAMING_SNAKE_CASE_ :     Union[str, Any]							=   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[int]							=   content_text_input.input_ids.shape[-1]
              SCREAMING_SNAKE_CASE_ :     Dict							=   self.tokenizer(['']       ,    padding='max_length'       ,    max_length=__lowerCAmelCase       ,    return_tensors='pt'    )
              SCREAMING_SNAKE_CASE_ :     Any							=   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(__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_ :     Union[str, Any]							=   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_ :     Tuple							=   text_embeddings.dtype
        if latents is None:
              if self.device.type == "mps":
                    # randn does not work reproducibly on mps
                    SCREAMING_SNAKE_CASE_ :     Union[str, Any]							=   torch.randn(__lowerCAmelCase       ,    generator=__lowerCAmelCase       ,    device='cpu'       ,    dtype=__lowerCAmelCase    ).to(
                        self.device    )
              else:
                    SCREAMING_SNAKE_CASE_ :     int							=   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_ :     Optional[int]							=   latents.to(self.device    )

        # scale the initial noise by the standard deviation required by the scheduler
        SCREAMING_SNAKE_CASE_ :     List[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_ :     Tuple							=   '''eta''' in set(inspect.signature(self.scheduler.step    ).parameters.keys()    )
        SCREAMING_SNAKE_CASE_ :     List[str]							=   {}
        if accepts_eta:
              SCREAMING_SNAKE_CASE_ :     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_ :     Any							=   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_ :     Dict							=   torch.cat([latents] * 2    ) if do_classifier_free_guidance else latents
                    SCREAMING_SNAKE_CASE_ :     Optional[int]							=   self.scheduler.scale_model_input(__lowerCAmelCase       ,    __lowerCAmelCase    )

                    # predict the noise residual
                    SCREAMING_SNAKE_CASE_ :     Optional[Any]							=   self.unet(__lowerCAmelCase       ,    __lowerCAmelCase       ,    encoder_hidden_states=__lowerCAmelCase    ).sample

                    # perform classifier free guidance
                    if do_classifier_free_guidance:
                          SCREAMING_SNAKE_CASE_ :     List[str]							=   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_ :     Union[str, Any]							=   (
                              text_embeddings.chunk(2    )[1] if do_classifier_free_guidance else text_embeddings
                          )
                          SCREAMING_SNAKE_CASE_ :     Any							=   self.cond_fn(
                              __lowerCAmelCase       ,    __lowerCAmelCase       ,    __lowerCAmelCase       ,    __lowerCAmelCase       ,    __lowerCAmelCase       ,    __lowerCAmelCase       ,    __lowerCAmelCase       ,    )

                    # compute the previous noisy sample x_t -> x_t-1
                    SCREAMING_SNAKE_CASE_ :     Optional[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_ :     Dict							=   1 / 0.18215 * latents
        SCREAMING_SNAKE_CASE_ :     Any							=   self.vae.decode(__lowerCAmelCase    ).sample

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

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

        if not return_dict:
              return (image, None)

        return StableDiffusionPipelineOutput(images=__lowerCAmelCase       ,    nsfw_content_detected=__lowerCAmelCase    )







'''simple docstring'''

import doctest
from collections import deque

import numpy as np



class 			UpperCAmelCase  :





       '''simple docstring'''



       def __init__(						self )  ->  None:
        lowercase__				:    str  =					[2, 1, 2, -1]
        lowercase__				:    str  =					[1, 2, 3, 4]


       def _lowerCAmelCase(						self )  ->  list[float]:
        lowercase__				:    Optional[Any]  =					len(self.first_signal )
        lowercase__				:    Union[str, Any]  =					len(self.second_signal )

        lowercase__				:    int  =					max(__lowerCAmelCase	,   __lowerCAmelCase )

        # create a zero matrix of max_length x max_length
        lowercase__				:    List[str]  =					[[0] * max_length for i in range(__lowerCAmelCase )]

        # fills the smaller signal with zeros to make both signals of same length
        if length_first_signal < length_second_signal:
         self.first_signal += [0] * (max_length - length_first_signal)
        elif length_first_signal > length_second_signal:
         self.second_signal += [0] * (max_length - length_second_signal)
        for i in range(__lowerCAmelCase ):
         lowercase__				:    int  =					deque(self.second_signal )
         rotated_signal.rotate(__lowerCAmelCase )
         for j, item in enumerate(__lowerCAmelCase ):
          matrix[i][j] += item

        # multiply the matrix with the first signal
        lowercase__				:    Optional[int]  =					np.matmul(np.transpose(__lowerCAmelCase )	,   np.transpose(self.first_signal ) )

        # rounding-off to two decimal places
        return [round(__lowerCAmelCase	,   2 ) for i in final_signal]


if __name__ == "__main__":
  doctest.testmod()

