






'''simple docstring'''






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

import torch
from transformers import (
    CLIPImageProcessor,
    CLIPTextModel,
    CLIPTokenizer,
    WhisperForConditionalGeneration,
    WhisperProcessor,
)

from diffusers import (
    AutoencoderKL,
    DDIMScheduler,
    DiffusionPipeline,
    LMSDiscreteScheduler,
    PNDMScheduler,
    UNetaDConditionModel,
)
from diffusers.pipelines.stable_diffusion.pipeline_stable_diffusion import StableDiffusionPipelineOutput
from diffusers.pipelines.stable_diffusion.safety_checker import StableDiffusionSafetyChecker
from diffusers.utils import logging


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




class a_    (    a__							):







   def __init__(self					,   __a					,   __a					,   __a					,   __a					,   __a					,   __a					,   __a					,   __a					,   __a					,   )			->      int:
      super().__init__()

      if safety_checker is None:
         logger.warning(
             F"""You have disabled the safety checker for {self.__class__} by passing `safety_checker=None`. Ensure"""
             ' that you abide to the conditions of the Stable Diffusion license and do not expose unfiltered'
             ' results in services or applications open to the public. Both the diffusers team and Hugging Face'
             ' strongly recommend to keep the safety filter enabled in all public facing circumstances, disabling'
             ' it only for use-cases that involve analyzing network behavior or auditing its results. For more'
             ' information, please have a look at https://github.com/huggingface/diffusers/pull/254 .')

      self.register_modules(
          speech_model=lowerCAmelCase__					,   speech_processor=lowerCAmelCase__					,   vae=lowerCAmelCase__					,   text_encoder=lowerCAmelCase__					,   tokenizer=lowerCAmelCase__					,   unet=lowerCAmelCase__					,   scheduler=lowerCAmelCase__					,   feature_extractor=lowerCAmelCase__					,   )







   def     SCREAMING_SNAKE_CASE__					(self					,   __a = "auto")			->      int:
      if slice_size == "auto":
         __snake_case				:      Any     							= self.unet.config.attention_head_dim // 2
      self.unet.set_attention_slice(lowerCAmelCase__)







   def     SCREAMING_SNAKE_CASE__					(self)			->      Tuple:
      self.enable_attention_slicing(lowerCAmelCase__)







   @torch.no_grad()
   def __call__(self					,   __a					,   __a=1_6_0_0_0					,   __a = 5_1_2					,   __a = 5_1_2					,   __a = 5_0					,   __a = 7.5					,   __a = None					,   __a = 1					,   __a = 0.0					,   __a = None					,   __a = None					,   __a = "pil"					,   __a = True					,   __a = None					,   __a = 1					,   **__a					,   )			->      Optional[Any]:
      __snake_case				:      Optional[int]     							= self.speech_processor.feature_extractor(
          lowerCAmelCase__					,   return_tensors='pt'					,   sampling_rate=lowerCAmelCase__).input_features.to(self.device)
      __snake_case				:      Optional[Any]     							= self.speech_model.generate(lowerCAmelCase__					,   max_length=4_8_0_0_0_0)

      __snake_case				:      Tuple     							= self.speech_processor.tokenizer.batch_decode(lowerCAmelCase__					,   skip_special_tokens=lowerCAmelCase__					,   normalize=lowerCAmelCase__)[
          0
      ]

      if isinstance(lowerCAmelCase__					,   lowerCAmelCase__):
         __snake_case				:      Optional[Any]     							= 1
      elif isinstance(lowerCAmelCase__					,   lowerCAmelCase__):
         __snake_case				:      Any     							= len(lowerCAmelCase__)
      else:
         raise ValueError(F"""`prompt` has to be of type `str` or `list` but is {type(lowerCAmelCase__)}""")

      if height % 8 != 0 or width % 8 != 0:
         raise ValueError(F"""`height` and `width` have to be divisible by 8 but are {height} and {width}.""")

      if (callback_steps is None) or (
          callback_steps is not None and (not isinstance(lowerCAmelCase__					,   lowerCAmelCase__) or callback_steps <= 0)
      ):
         raise ValueError(
             F"""`callback_steps` has to be a positive integer but is {callback_steps} of type"""
             F""" {type(lowerCAmelCase__)}.""")

      # get prompt text embeddings
      __snake_case				:      Union[str, Any]     							= self.tokenizer(
          lowerCAmelCase__					,   padding='max_length'					,   max_length=self.tokenizer.model_max_length					,   return_tensors='pt'					,   )
      __snake_case				:      List[Any]     							= text_inputs.input_ids

      if text_input_ids.shape[-1] > self.tokenizer.model_max_length:
         __snake_case				:      Any     							= self.tokenizer.batch_decode(text_input_ids[:, self.tokenizer.model_max_length :])
         logger.warning(
             'The following part of your input was truncated because CLIP can only handle sequences up to'
             F""" {self.tokenizer.model_max_length} tokens: {removed_text}""")
         __snake_case				:      Optional[Any]     							= text_input_ids[:, : self.tokenizer.model_max_length]
      __snake_case				:      int     							= self.text_encoder(text_input_ids.to(self.device))[0]

      # duplicate text embeddings for each generation per prompt, using mps friendly method
      __snake_case				:      Dict     							= text_embeddings.shape
      __snake_case				:      Any     							= text_embeddings.repeat(1					,   lowerCAmelCase__					,   1)
      __snake_case				:      Optional[Any]     							= text_embeddings.view(bs_embed * num_images_per_prompt					,   lowerCAmelCase__					,   -1)

      # here `guidance_scale` is defined analog to the guidance weight `w` of equation (2)
      # of the Imagen paper: https://arxiv.org/pdf/2205.11487.pdf . `guidance_scale = 1`
      # corresponds to doing no classifier free guidance.
      __snake_case				:      Optional[int]     							= guidance_scale > 1.0
      # get unconditional embeddings for classifier free guidance
      if do_classifier_free_guidance:
         __snake_case				:      List[str]
         if negative_prompt is None:
            __snake_case				:      Any     							= [""] * batch_size
         elif type(lowerCAmelCase__) is not type(lowerCAmelCase__):
            raise TypeError(
                F"""`negative_prompt` should be the same type to `prompt`, but got {type(lowerCAmelCase__)} !="""
                F""" {type(lowerCAmelCase__)}.""")
         elif isinstance(lowerCAmelCase__					,   lowerCAmelCase__):
            __snake_case				:      Optional[Any]     							= [negative_prompt]
         elif batch_size != len(lowerCAmelCase__):
            raise ValueError(
                F"""`negative_prompt`: {negative_prompt} has batch size {len(lowerCAmelCase__)}, but `prompt`:"""
                F""" {prompt} has batch size {batch_size}. Please make sure that passed `negative_prompt` matches"""
                ' the batch size of `prompt`.')
         else:
            __snake_case				:      Optional[int]     							= negative_prompt

         __snake_case				:      Dict     							= text_input_ids.shape[-1]
         __snake_case				:      Any     							= self.tokenizer(
             lowerCAmelCase__					,   padding='max_length'					,   max_length=lowerCAmelCase__					,   truncation=lowerCAmelCase__					,   return_tensors='pt'					,   )
         __snake_case				:      Optional[Any]     							= self.text_encoder(uncond_input.input_ids.to(self.device))[0]

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

         # For classifier free guidance, we need to do two forward passes.
         # Here we concatenate the unconditional and text embeddings into a single batch
         # to avoid doing two forward passes
         __snake_case				:      Optional[int]     							= torch.cat([uncond_embeddings, text_embeddings])

      # get the initial random noise unless the user supplied it

      # Unlike in other pipelines, latents need to be generated in the target device
      # for 1-to-1 results reproducibility with the CompVis implementation.
      # However this currently doesn't work in `mps`.
      __snake_case				:      List[Any]     							= (batch_size * num_images_per_prompt, self.unet.config.in_channels, height // 8, width // 8)
      __snake_case				:      Tuple     							= text_embeddings.dtype
      if latents is None:
         if self.device.type == "mps":
            # randn does not exist on mps
            __snake_case				:      Union[str, Any]     							= torch.randn(lowerCAmelCase__					,   generator=lowerCAmelCase__					,   device='cpu'					,   dtype=lowerCAmelCase__).to(
                self.device)
         else:
            __snake_case				:      Optional[Any]     							= 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}""")
         __snake_case				:      str     							= latents.to(self.device)

      # set timesteps
      self.scheduler.set_timesteps(lowerCAmelCase__)

      # Some schedulers like PNDM have timesteps as arrays
      # It's more optimized to move all timesteps to correct device beforehand
      __snake_case				:      Optional[Any]     							= self.scheduler.timesteps.to(self.device)

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

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

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

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

         # compute the previous noisy sample x_t -> x_t-1
         __snake_case				:      List[str]     							= self.scheduler.step(lowerCAmelCase__					,   lowerCAmelCase__					,   lowerCAmelCase__					,   **lowerCAmelCase__).prev_sample

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

      __snake_case				:      Optional[Any]     							= 1 / 0.18_215 * latents
      __snake_case				:      Union[str, Any]     							= self.vae.decode(lowerCAmelCase__).sample

      __snake_case				:      int     							= (image / 2 + 0.5).clamp(0					,   1)

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

      if output_type == "pil":
         __snake_case				:      Optional[int]     							= self.numpy_to_pil(lowerCAmelCase__)

      if not return_dict:
         return image

      return StableDiffusionPipelineOutput(images=lowerCAmelCase__					,   nsfw_content_detected=lowerCAmelCase__)







'''simple docstring'''






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 (
    center_crop,
    get_resize_output_image_size,
    normalize,
    rescale,
    resize,
    to_channel_dimension_format,
)
from ...image_utils import (
    IMAGENET_STANDARD_MEAN,
    IMAGENET_STANDARD_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


__A    	=   logging.get_logger(__name__)




class a_    (    UpperCamelCase_							):
   _snake_case        =							["""pixel_values"""]







   def __init__(self					,   __a = True					,   __a = None					,   __a = None					,   __a = PILImageResampling.BILINEAR					,   __a = True					,   __a = 1 / 2_5_5					,   __a = True					,   __a = None					,   __a = None					,   **__a					,   )			->      None:
      """simple docstring"""






      super().__init__(**__a)
      __snake_case				:      Tuple     							= size if size is not None else {'shortest_edge': 3_8_4}
      __snake_case				:      List[Any]     							= get_size_dict(__a					,   default_to_square=__a)

      __snake_case				:      int     							= do_resize
      __snake_case				:      List[str]     							= size
      # Default value set here for backwards compatibility where the value in config is None
      __snake_case				:      Any     							= crop_pct if crop_pct is not None else 2_2_4 / 2_5_6
      __snake_case				:      Tuple     							= resample
      __snake_case				:      Dict     							= do_rescale
      __snake_case				:      Any     							= rescale_factor
      __snake_case				:      str     							= do_normalize
      __snake_case				:      Union[str, Any]     							= image_mean if image_mean is not None else IMAGENET_STANDARD_MEAN
      __snake_case				:      Dict     							= image_std if image_std is not None else IMAGENET_STANDARD_STD







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






      __snake_case				:      Dict     							= get_size_dict(__a					,   default_to_square=__a)
      if "shortest_edge" not in size:
         raise ValueError(F"""Size dictionary must contain 'shortest_edge' key. Got {size.keys()}""")
      __snake_case				:      List[str]     							= size['shortest_edge']

      if shortest_edge < 3_8_4:
         # maintain same ratio, resizing shortest edge to shortest_edge/crop_pct
         __snake_case				:      Any     							= int(shortest_edge / crop_pct)
         __snake_case				:      Any     							= get_resize_output_image_size(__a					,   size=__a					,   default_to_square=__a)
         __snake_case				:      int     							= resize(image=__a					,   size=__a					,   resample=__a					,   data_format=__a					,   **__a)
         # then crop to (shortest_edge, shortest_edge)
         return center_crop(image=__a					,   size=(shortest_edge, shortest_edge)					,   data_format=__a					,   **__a)
      else:
         # warping (no cropping) when evaluated at 384 or larger
         return resize(
             __a					,   size=(shortest_edge, shortest_edge)					,   resample=__a					,   data_format=__a					,   **__a)







   def     SCREAMING_SNAKE_CASE__					(self					,   __a					,   __a					,   __a = None					,   **__a					,   )			->      Any:
      """simple docstring"""






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







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






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







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






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

      __snake_case				:      List[str]     							= size if size is not None else self.size
      __snake_case				:      Any     							= get_size_dict(__a					,   default_to_square=__a)

      __snake_case				:      Dict     							= make_list_of_images(__a)

      if not valid_images(__a):
         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_resize and size["shortest_edge"] < 3_8_4 and crop_pct is None:
         raise ValueError('crop_pct must be specified if size < 384.')

      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.')

      # All transformations expect numpy arrays.
      __snake_case				:      Tuple     							= [to_numpy_array(__a) for image in images]

      if do_resize:
         __snake_case				:      Optional[int]     							= [self.resize(image=__a					,   size=__a					,   crop_pct=__a					,   resample=__a) for image in images]

      if do_rescale:
         __snake_case				:      Optional[int]     							= [self.rescale(image=__a					,   scale=__a) for image in images]

      if do_normalize:
         __snake_case				:      Any     							= [self.normalize(image=__a					,   mean=__a					,   std=__a) for image in images]

      __snake_case				:      Dict     							= [to_channel_dimension_format(__a					,   __a) for image in images]

      __snake_case				:      Union[str, Any]     							= {'pixel_values': images}
      return BatchFeature(data=__a					,   tensor_type=__a)