
"""simple docstring"""



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 					lowerCamelCase   (     _snake_case    ,_snake_case    ,_snake_case ):
    if isinstance(_snake_case    ,torch.Tensor ):
        return image
    elif isinstance(_snake_case    ,PIL.Image.Image ):
        UpperCAmelCase__	:      List[str]     							=					[image]

    if isinstance(image[0]    ,PIL.Image.Image ):
        UpperCAmelCase__	:      Optional[int]     							=					[np.array(i.resize((w, h)    ,resample=PIL_INTERPOLATION['lanczos'] ) )[None, :] for i in image]
        UpperCAmelCase__	:      Dict     							=					np.concatenate(_snake_case    ,axis=0 )
        UpperCAmelCase__	:      Union[str, Any]     							=					np.array(_snake_case ).astype(np.floataa ) / 255.0
        UpperCAmelCase__	:      int     							=					image.transpose(0    ,3    ,1    ,2 )
        UpperCAmelCase__	:      List[Any]     							=					2.0 * image - 1.0
        UpperCAmelCase__	:      Tuple     							=					torch.from_numpy(_snake_case )
    elif isinstance(image[0]    ,torch.Tensor ):
        UpperCAmelCase__	:      str     							=					torch.cat(_snake_case    ,dim=0 )
    return image
def 					lowerCamelCase   (     _snake_case    ,_snake_case    ,_snake_case    ,_snake_case=0.9995 ):
    if not isinstance(_snake_case    ,np.ndarray ):
        UpperCAmelCase__	:      Dict     							=					True
        UpperCAmelCase__	:      Dict     							=					va.device
        UpperCAmelCase__	:      List[str]     							=					va.cpu().numpy()
        UpperCAmelCase__	:      str     							=					va.cpu().numpy()

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

    if inputs_are_torch:
        UpperCAmelCase__	:      Optional[int]     							=					torch.from_numpy(_snake_case ).to(_snake_case )

    return va
def 					lowerCamelCase   (     _snake_case    ,_snake_case ):
    UpperCAmelCase__	:      Optional[Any]     							=					F.normalize(_snake_case    ,dim=-1 )
    UpperCAmelCase__	:      Union[str, Any]     							=					F.normalize(_snake_case    ,dim=-1 )
    return (x - y).norm(dim=-1 ).div(2 ).arcsin().pow(2 ).mul(2 )
def 					lowerCamelCase   (     _snake_case    ,_snake_case ):
    for param in model.parameters():
        UpperCAmelCase__	:      int     							=					value





class 				a			(					_UpperCamelCase      ):



      def __init__(   self     ,	UpperCamelCase_     ,	UpperCamelCase_     ,	UpperCamelCase_     ,	UpperCamelCase_     ,	UpperCamelCase_     ,	UpperCamelCase_     ,	UpperCamelCase_     ,	UpperCamelCase_=None     ,	UpperCamelCase_=None     ,	UpperCamelCase_=None     ,	):
          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_     ,	)
          UpperCAmelCase__	:      Union[str, Any]     							=					(
              feature_extractor.size
              if isinstance(feature_extractor.size     ,	UpperCamelCase_      )
              else feature_extractor.size["shortest_edge"]
          )
          UpperCAmelCase__	:      Optional[Any]     							=					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  __snake_case							(   self     ,	UpperCamelCase_ = "auto"      ):





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



      def  __snake_case							(   self      ):
          self.enable_attention_slicing(UpperCamelCase_      )



      def  __snake_case							(   self      ):
          set_requires_grad(self.vae     ,	UpperCamelCase_      )



      def  __snake_case							(   self      ):
          set_requires_grad(self.vae     ,	UpperCamelCase_      )



      def  __snake_case							(   self      ):
          set_requires_grad(self.unet     ,	UpperCamelCase_      )



      def  __snake_case							(   self      ):
          set_requires_grad(self.unet     ,	UpperCamelCase_      )



      def  __snake_case							(   self     ,	UpperCamelCase_     ,	UpperCamelCase_     ,	UpperCamelCase_      ):
          UpperCAmelCase__	:      Dict     							=					min(int(num_inference_steps * strength      )     ,	UpperCamelCase_      )

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

          return timesteps, num_inference_steps - t_start



      def  __snake_case							(   self     ,	UpperCamelCase_     ,	UpperCamelCase_     ,	UpperCamelCase_     ,	UpperCamelCase_     ,	UpperCamelCase_     ,	UpperCamelCase_=None      ):





          if not isinstance(UpperCamelCase_     ,	torch.Tensor      ):
              raise ValueError(F'''`image` has to be of type `torch.Tensor` but is {type(UpperCamelCase_      )}'''      )

          UpperCAmelCase__	:      List[str]     							=					image.to(device=UpperCamelCase_     ,	dtype=UpperCamelCase_      )

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

          # Hardcode 0.18215 because stable-diffusion-2-base has not self.vae.config.scaling_factor
          UpperCAmelCase__	:      Optional[int]     							=					0.18215 * init_latents
          UpperCAmelCase__	:      Any     							=					init_latents.repeat_interleave(UpperCamelCase_     ,	dim=0      )

          UpperCAmelCase__	:      Optional[int]     							=					randn_tensor(init_latents.shape     ,	generator=UpperCamelCase_     ,	device=UpperCamelCase_     ,	dtype=UpperCamelCase_      )

          # get latents
          UpperCAmelCase__	:      Tuple     							=					self.scheduler.add_noise(UpperCamelCase_     ,	UpperCamelCase_     ,	UpperCamelCase_      )
          UpperCAmelCase__	:      Any     							=					init_latents

          return latents



      def  __snake_case							(   self     ,	UpperCamelCase_      ):
          UpperCAmelCase__	:      Dict     							=					self.coca_transform(UpperCamelCase_      ).unsqueeze(0      )
          with torch.no_grad(), torch.cuda.amp.autocast():
              UpperCAmelCase__	:      Dict     							=					self.coca_model.generate(transformed_image.to(device=self.device     ,	dtype=self.coca_model.dtype      )      )
          UpperCAmelCase__	:      List[Any]     							=					self.coca_tokenizer.decode(generated[0].cpu().numpy()      )
          return generated.split('<end_of_text>'      )[0].replace('<start_of_text>'     ,	''      ).rstrip(' .,'      )



      def  __snake_case							(   self     ,	UpperCamelCase_     ,	UpperCamelCase_      ):
          UpperCAmelCase__	:      Optional[int]     							=					self.feature_extractor.preprocess(UpperCamelCase_      )
          UpperCAmelCase__	:      Optional[Any]     							=					torch.from_numpy(clip_image_input['pixel_values'][0]      ).unsqueeze(0      ).to(self.device      ).half()
          UpperCAmelCase__	:      Tuple     							=					self.clip_model.get_image_features(UpperCamelCase_      )
          UpperCAmelCase__	:      Dict     							=					image_embeddings_clip / image_embeddings_clip.norm(p=2     ,	dim=-1     ,	keepdim=UpperCamelCase_      )
          UpperCAmelCase__	:      Optional[int]     							=					image_embeddings_clip.repeat_interleave(UpperCamelCase_     ,	dim=0      )
          return image_embeddings_clip



      @torch.enable_grad()
      def  __snake_case							(   self     ,	UpperCamelCase_     ,	UpperCamelCase_     ,	UpperCamelCase_     ,	UpperCamelCase_     ,	UpperCamelCase_     ,	UpperCamelCase_     ,	UpperCamelCase_     ,	):
          UpperCAmelCase__	:      Any     							=					latents.detach().requires_grad_()

          UpperCAmelCase__	:      Any     							=					self.scheduler.scale_model_input(UpperCamelCase_     ,	UpperCamelCase_      )

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

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

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

          UpperCAmelCase__	:      Any     							=					transforms.Resize(self.feature_extractor_size      )(UpperCamelCase_      )
          UpperCAmelCase__	:      Optional[int]     							=					self.normalize(UpperCamelCase_      ).to(latents.dtype      )

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

          UpperCAmelCase__	:      Dict     							=					spherical_dist_loss(UpperCamelCase_     ,	UpperCamelCase_      ).mean() * clip_guidance_scale

          UpperCAmelCase__	:      int     							=					-torch.autograd.grad(UpperCamelCase_     ,	UpperCamelCase_      )[0]

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



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





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

          UpperCAmelCase__	:      Tuple     							=					[
              ("model", self.coca_model is None),
              ("tokenizer", self.coca_tokenizer is None),
              ("transform", self.coca_transform is None),
          ]
          UpperCAmelCase__	:      Any     							=					[x[0] for x in coca_is_none if x[1]]
          UpperCAmelCase__	:      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.'''      )
              UpperCAmelCase__	:      Tuple     							=					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.'''      )
              UpperCAmelCase__	:      List[Any]     							=					self.get_image_description(UpperCamelCase_      )

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

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

          UpperCAmelCase__	:      Optional[int]     							=					slerp(UpperCamelCase_     ,	UpperCamelCase_     ,	UpperCamelCase_      )

          # duplicate text embeddings for each generation per prompt
          UpperCAmelCase__	:      List[Any]     							=					text_embeddings.repeat_interleave(UpperCamelCase_     ,	dim=0      )

          # set timesteps
          UpperCAmelCase__	:      Optional[Any]     							=					"offset" in set(inspect.signature(self.scheduler.set_timesteps      ).parameters.keys()      )
          UpperCAmelCase__	:      Tuple     							=					{}
          if accepts_offset:
              UpperCAmelCase__	:      Union[str, Any]     							=					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      )

          UpperCAmelCase__	:      Optional[Any]     							=					self.get_timesteps(UpperCamelCase_     ,	UpperCamelCase_     ,	self.device      )
          UpperCAmelCase__	:      List[str]     							=					timesteps[:1].repeat(UpperCamelCase_      )

          # Preprocess image
          UpperCAmelCase__	:      Tuple     							=					preprocess(UpperCamelCase_     ,	UpperCamelCase_     ,	UpperCamelCase_      )
          UpperCAmelCase__	:      str     							=					self.prepare_latents(
              UpperCamelCase_     ,	UpperCamelCase_     ,	UpperCamelCase_     ,	text_embeddings.dtype     ,	self.device     ,	UpperCamelCase_      )

          UpperCAmelCase__	:      List[str]     							=					preprocess(UpperCamelCase_     ,	UpperCamelCase_     ,	UpperCamelCase_      )
          UpperCAmelCase__	:      Tuple     							=					self.prepare_latents(
              UpperCamelCase_     ,	UpperCamelCase_     ,	UpperCamelCase_     ,	text_embeddings.dtype     ,	self.device     ,	UpperCamelCase_      )

          UpperCAmelCase__	:      Union[str, Any]     							=					slerp(UpperCamelCase_     ,	UpperCamelCase_     ,	UpperCamelCase_      )

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

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

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

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

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

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

                  # perform clip guidance
                  if clip_guidance_scale > 0:
                      UpperCAmelCase__	:      Dict     							=					(
                          text_embeddings.chunk(2      )[1] if do_classifier_free_guidance else text_embeddings
                      )
                      UpperCAmelCase__	:      List[str]     							=					self.cond_fn(
                          UpperCamelCase_     ,	UpperCamelCase_     ,	UpperCamelCase_     ,	UpperCamelCase_     ,	UpperCamelCase_     ,	UpperCamelCase_     ,	UpperCamelCase_     ,	)

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

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

          if output_type == "pil":
              UpperCAmelCase__	:      str     							=					self.numpy_to_pil(UpperCamelCase_      )

          if not return_dict:
              return (image, None)

          return StableDiffusionPipelineOutput(images=UpperCamelCase_     ,	nsfw_content_detected=UpperCamelCase_      )








from typing import Dict, List, Optional, Union

import numpy as np

from ...image_processing_utils import BaseImageProcessor, BatchFeature, get_size_dict
from ...image_transforms import convert_to_rgb, normalize, rescale, resize, to_channel_dimension_format
from ...image_utils import (
    OPENAI_CLIP_MEAN,
    OPENAI_CLIP_STD,
    ChannelDimension,
    ImageInput,
    PILImageResampling,
    make_list_of_images,
    to_numpy_array,
    valid_images,
)
from ...utils import TensorType, is_vision_available, logging


if is_vision_available():
  import PIL


SCREAMING_SNAKE_CASE__							:      List[str]         =    logging.get_logger(__name__)







class 	__lowerCAmelCase      (       _UpperCamelCase	):
   _UpperCamelCase			:    int  		=					["""pixel_values"""]





   def __init__(			self       ,      snake_case = True       ,      snake_case = None       ,      snake_case = PILImageResampling.BICUBIC       ,      snake_case = True       ,      snake_case = 1 / 255       ,      snake_case = True       ,      snake_case = None       ,      snake_case = None       ,      snake_case = True       ,      **snake_case       ,      )					->	None:
      """simple docstring"""





      super().__init__(**snake_case   )
      a__						:    List[str]   		=	size if size is not None else {"height": 384, "width": 384}
      a__						:    List[str]   		=	get_size_dict(snake_case       ,      default_to_square=snake_case   )

      a__						:    Any   		=	do_resize
      a__						:    Optional[int]   		=	size
      a__						:    int   		=	resample
      a__						:    Optional[Any]   		=	do_rescale
      a__						:    Dict   		=	rescale_factor
      a__						:    Dict   		=	do_normalize
      a__						:    Any   		=	image_mean if image_mean is not None else OPENAI_CLIP_MEAN
      a__						:    List[str]   		=	image_std if image_std is not None else OPENAI_CLIP_STD
      a__						:    str   		=	do_convert_rgb





   def 						_snake_case			(			self       ,      snake_case       ,      snake_case       ,      snake_case = PILImageResampling.BICUBIC       ,      snake_case = None       ,      **snake_case       ,      )					->	np.ndarray:
      """simple docstring"""





      a__						:    Optional[int]   		=	get_size_dict(snake_case       ,      default_to_square=snake_case   )
      if "height" not in size or "width" not in size:
         raise ValueError(F"""The `size` dictionary must contain the keys `height` and `width`. Got {size.keys()}"""   )
      a__						:    Optional[int]   		=	(size["height"], size["width"])
      return resize(snake_case       ,      size=snake_case       ,      resample=snake_case       ,      data_format=snake_case       ,      **snake_case   )





   def 						_snake_case			(			self       ,      snake_case       ,      snake_case       ,      snake_case = None       ,      **snake_case       ,      )					->	List[str]:
      """simple docstring"""





      return rescale(snake_case       ,      scale=snake_case       ,      data_format=snake_case       ,      **snake_case   )





   def 						_snake_case			(			self       ,      snake_case       ,      snake_case       ,      snake_case       ,      snake_case = None       ,      **snake_case       ,      )					->	np.ndarray:
      """simple docstring"""





      return normalize(snake_case       ,      mean=snake_case       ,      std=snake_case       ,      data_format=snake_case       ,      **snake_case   )





   def 						_snake_case			(			self       ,      snake_case       ,      snake_case = None       ,      snake_case = None       ,      snake_case = None       ,      snake_case = None       ,      snake_case = None       ,      snake_case = None       ,      snake_case = None       ,      snake_case = None       ,      snake_case = None       ,      snake_case = None       ,      snake_case = ChannelDimension.FIRST       ,      **snake_case       ,      )					->	PIL.Image.Image:
      """simple docstring"""





      a__						:    List[str]   		=	do_resize if do_resize is not None else self.do_resize
      a__						:    Optional[Any]   		=	resample if resample is not None else self.resample
      a__						:    int   		=	do_rescale if do_rescale is not None else self.do_rescale
      a__						:    int   		=	rescale_factor if rescale_factor is not None else self.rescale_factor
      a__						:    Union[str, Any]   		=	do_normalize if do_normalize is not None else self.do_normalize
      a__						:    List[Any]   		=	image_mean if image_mean is not None else self.image_mean
      a__						:    int   		=	image_std if image_std is not None else self.image_std
      a__						:    Tuple   		=	do_convert_rgb if do_convert_rgb is not None else self.do_convert_rgb

      a__						:    Tuple   		=	size if size is not None else self.size
      a__						:    Any   		=	get_size_dict(snake_case       ,      default_to_square=snake_case   )

      a__						:    str   		=	make_list_of_images(snake_case   )

      if not valid_images(snake_case   ):
         raise ValueError(
             "Invalid image type. Must be of type PIL.Image.Image, numpy.ndarray, "
             "torch.Tensor, tf.Tensor or jax.ndarray."   )

      if do_resize and size is None or resample is None:
         raise ValueError("Size and resample must be specified if do_resize is True."   )

      if do_rescale and rescale_factor is None:
         raise ValueError("Rescale factor must be specified if do_rescale is True."   )

      if do_normalize and (image_mean is None or image_std is None):
         raise ValueError("Image mean and std must be specified if do_normalize is True."   )

      # PIL RGBA images are converted to RGB
      if do_convert_rgb:
         a__						:    List[str]   		=	[convert_to_rgb(snake_case   ) for image in images]

      # All transformations expect numpy arrays.
      a__						:    Dict   		=	[to_numpy_array(snake_case   ) for image in images]

      if do_resize:
         a__						:    Union[str, Any]   		=	[self.resize(image=snake_case       ,      size=snake_case       ,      resample=snake_case   ) for image in images]

      if do_rescale:
         a__						:    Tuple   		=	[self.rescale(image=snake_case       ,      scale=snake_case   ) for image in images]

      if do_normalize:
         a__						:    List[str]   		=	[self.normalize(image=snake_case       ,      mean=snake_case       ,      std=snake_case   ) for image in images]

      a__						:    Union[str, Any]   		=	[to_channel_dimension_format(snake_case       ,      snake_case   ) for image in images]

      a__						:    List[Any]   		=	BatchFeature(data={"pixel_values": images}       ,      tensor_type=snake_case   )

      return encoded_outputs

