






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


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

class   _UpperCAmelCase     (	snake_case_		):




   """simple docstring"""

   def __init__(						self					:      List[Any] , __UpperCAmelCase					:      WhisperForConditionalGeneration , __UpperCAmelCase					:      WhisperProcessor , __UpperCAmelCase					:      AutoencoderKL , __UpperCAmelCase					:      CLIPTextModel , __UpperCAmelCase					:      CLIPTokenizer , __UpperCAmelCase					:      UNetaDConditionModel , __UpperCAmelCase					:      Union[DDIMScheduler, PNDMScheduler, LMSDiscreteScheduler] , __UpperCAmelCase					:      StableDiffusionSafetyChecker , __UpperCAmelCase					:      CLIPImageProcessor , ):





         '''simple docstring'''





         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=a_ , speech_processor=a_ , vae=a_ , text_encoder=a_ , tokenizer=a_ , unet=a_ , scheduler=a_ , feature_extractor=a_ , )

   def 		lowerCAmelCase					(						self					:      List[str] , __UpperCAmelCase					:      Optional[Union[str, int]] = "auto"						):





         '''simple docstring'''





         if slice_size == "auto":
               _A     	=      self.unet.config.attention_head_dim // 2
         self.unet.set_attention_slice(a_						)

   def 		lowerCAmelCase					(						self					:      List[str]						):





         '''simple docstring'''





         self.enable_attention_slicing(a_						)

   @torch.no_grad()
   def __call__(						self					:      List[Any] , __UpperCAmelCase					:      List[str] , __UpperCAmelCase					:      Optional[Any]=16000 , __UpperCAmelCase					:      int = 512 , __UpperCAmelCase					:      int = 512 , __UpperCAmelCase					:      int = 50 , __UpperCAmelCase					:      float = 7.5 , __UpperCAmelCase					:      Optional[Union[str, List[str]]] = None , __UpperCAmelCase					:      Optional[int] = 1 , __UpperCAmelCase					:      float = 0.0 , __UpperCAmelCase					:      Optional[torch.Generator] = None , __UpperCAmelCase					:      Optional[torch.FloatTensor] = None , __UpperCAmelCase					:      Optional[str] = "pil" , __UpperCAmelCase					:      bool = True , __UpperCAmelCase					:      Optional[Callable[[int, int, torch.FloatTensor], None]] = None , __UpperCAmelCase					:      int = 1 , **__UpperCAmelCase					:      Any , ):





         '''simple docstring'''





         _A     	=      self.speech_processor.feature_extractor(
             a_ , return_tensors="pt" , sampling_rate=a_						).input_features.to(self.device						)
         _A     	=      self.speech_model.generate(a_ , max_length=480000						)

         _A     	=      self.speech_processor.tokenizer.batch_decode(a_ , skip_special_tokens=a_ , normalize=a_						)[
             0
         ]

         if isinstance(a_ , a_						):
               _A     	=      1
         elif isinstance(a_ , a_						):
               _A     	=      len(a_						)
         else:
               raise ValueError(f'''`prompt` has to be of type `str` or `list` but is {type(a_						)}'''						)

         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(a_ , a_						) or callback_steps <= 0)
         ):
               raise ValueError(
                   f'''`callback_steps` has to be a positive integer but is {callback_steps} of type'''
                   f''' {type(a_						)}.'''						)

         # get prompt text embeddings
         _A     	=      self.tokenizer(
             a_ , padding="max_length" , max_length=self.tokenizer.model_max_length , return_tensors="pt" , )
         _A     	=      text_inputs.input_ids

         if text_input_ids.shape[-1] > self.tokenizer.model_max_length:
               _A     	=      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}'''						)
               _A     	=      text_input_ids[:, : self.tokenizer.model_max_length]
         _A     	=      self.text_encoder(text_input_ids.to(self.device						)						)[0]

         # duplicate text embeddings for each generation per prompt, using mps friendly method
         _A     	=      text_embeddings.shape
         _A     	=      text_embeddings.repeat(1 , a_ , 1						)
         _A     	=      text_embeddings.view(bs_embed * num_images_per_prompt , a_ , -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.
         _A     	=      guidance_scale > 1.0
         # get unconditional embeddings for classifier free guidance
         if do_classifier_free_guidance:
               _A     	=      42
               if negative_prompt is None:
                     _A     	=      [''] * batch_size
               elif type(a_						) is not type(a_						):
                     raise TypeError(
                         f'''`negative_prompt` should be the same type to `prompt`, but got {type(a_						)} !='''
                         f''' {type(a_						)}.'''						)
               elif isinstance(a_ , a_						):
                     _A     	=      [negative_prompt]
               elif batch_size != len(a_						):
                     raise ValueError(
                         f'''`negative_prompt`: {negative_prompt} has batch size {len(a_						)}, but `prompt`:'''
                         f''' {prompt} has batch size {batch_size}. Please make sure that passed `negative_prompt` matches'''
                         " the batch size of `prompt`."						)
               else:
                     _A     	=      negative_prompt

               _A     	=      text_input_ids.shape[-1]
               _A     	=      self.tokenizer(
                   a_ , padding="max_length" , max_length=a_ , truncation=a_ , return_tensors="pt" , )
               _A     	=      self.text_encoder(uncond_input.input_ids.to(self.device						)						)[0]

               # duplicate unconditional embeddings for each generation per prompt, using mps friendly method
               _A     	=      uncond_embeddings.shape[1]
               _A     	=      uncond_embeddings.repeat(1 , a_ , 1						)
               _A     	=      uncond_embeddings.view(batch_size * num_images_per_prompt , a_ , -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
               _A     	=      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`.
         _A     	=      (batch_size * num_images_per_prompt, self.unet.config.in_channels, height // 8, width // 8)
         _A     	=      text_embeddings.dtype
         if latents is None:
               if self.device.type == "mps":
                     # randn does not exist on mps
                     _A     	=      torch.randn(a_ , generator=a_ , device="cpu" , dtype=a_						).to(
                         self.device						)
               else:
                     _A     	=      torch.randn(a_ , generator=a_ , device=self.device , dtype=a_						)
         else:
               if latents.shape != latents_shape:
                     raise ValueError(f'''Unexpected latents shape, got {latents.shape}, expected {latents_shape}'''						)
               _A     	=      latents.to(self.device						)

         # set timesteps
         self.scheduler.set_timesteps(a_						)

         # Some schedulers like PNDM have timesteps as arrays
         # It's more optimized to move all timesteps to correct device beforehand
         _A     	=      self.scheduler.timesteps.to(self.device						)

         # scale the initial noise by the standard deviation required by the scheduler
         _A     	=      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]
         _A     	=      'eta' in set(inspect.signature(self.scheduler.step						).parameters.keys()						)
         _A     	=      {}
         if accepts_eta:
               _A     	=      eta

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

               # predict the noise residual
               _A     	=      self.unet(a_ , a_ , encoder_hidden_states=a_						).sample

               # perform guidance
               if do_classifier_free_guidance:
                     _A     	=      noise_pred.chunk(2						)
                     _A     	=      noise_pred_uncond + guidance_scale * (noise_pred_text - noise_pred_uncond)

               # compute the previous noisy sample x_t -> x_t-1
               _A     	=      self.scheduler.step(a_ , a_ , a_ , **a_						).prev_sample

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

         _A     	=      1 / 0.18215 * latents
         _A     	=      self.vae.decode(a_						).sample

         _A     	=      (image / 2 + 0.5).clamp(0 , 1						)

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

         if output_type == "pil":
               _A     	=      self.numpy_to_pil(a_						)

         if not return_dict:
               return image

         return StableDiffusionPipelineOutput(images=a_ , nsfw_content_detected=a_						)










'''simple docstring'''

from diffusers.utils.testing_utils import require_onnxruntime






@require_onnxruntime
class 						a_		:
							pass




