




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				:		str     =   logging.get_logger(__name__)  # pylint: disable=invalid-name







class   A(						UpperCamelCase   ):



 '''simple docstring'''


 def __init__(				self					:  str						,							A_					:  WhisperForConditionalGeneration						,							A_					:  WhisperProcessor						,							A_					:  AutoencoderKL						,							A_					:  CLIPTextModel						,							A_					:  CLIPTokenizer						,							A_					:  UNetaDConditionModel						,							A_					:  Union[DDIMScheduler, PNDMScheduler, LMSDiscreteScheduler]						,							A_					:  StableDiffusionSafetyChecker						,							A_					:  CLIPImageProcessor						,							)   ->	Union[str, Any]:

       """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 				a__ (				self					:  Optional[int]						,							A_					:  Optional[Union[str, int]] = "auto"  )   ->	Any:

       """simple docstring"""






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

 def 				a__ (				self					:  List[Any]  )   ->	List[str]:

       """simple docstring"""






       self.enable_attention_slicing(A_  )





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

       """simple docstring"""






       lowerCamelCase_		     =    self.speech_processor.feature_extractor(
           A_						,							return_tensors='pt'						,							sampling_rate=A_  ).input_features.to(self.device  )
       lowerCamelCase_		     =    self.speech_model.generate(A_						,							max_length=480000  )

       lowerCamelCase_		     =    self.speech_processor.tokenizer.batch_decode(A_						,							skip_special_tokens=A_						,							normalize=A_  )[
           0
       ]

       if isinstance(A_						,							A_  ):
             lowerCamelCase_		     =    1
       elif isinstance(A_						,							A_  ):
             lowerCamelCase_		     =    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
       lowerCamelCase_		     =    self.tokenizer(
           A_						,							padding='max_length'						,							max_length=self.tokenizer.model_max_length						,							return_tensors='pt'						,							)
       lowerCamelCase_		     =    text_inputs.input_ids

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

       # duplicate text embeddings for each generation per prompt, using mps friendly method
       lowerCamelCase_     ,				lowerCamelCase_     ,				lowerCamelCase_		     =    text_embeddings.shape
       lowerCamelCase_		     =    text_embeddings.repeat(1						,							A_						,							1  )
       lowerCamelCase_		     =    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.
       lowerCamelCase_		     =    guidance_scale > 1.0
       # get unconditional embeddings for classifier free guidance
       if do_classifier_free_guidance:
             lowerCamelCase_		     =    42
             if negative_prompt is None:
                   lowerCamelCase_		     =    [''] * 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_  ):
                   lowerCamelCase_		     =    [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:
                   lowerCamelCase_		     =    negative_prompt

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

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

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

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

             # predict the noise residual
             lowerCamelCase_		     =    self.unet(A_						,							A_						,							encoder_hidden_states=A_  ).sample

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

             # compute the previous noisy sample x_t -> x_t-1
             lowerCamelCase_		     =    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_  )

       lowerCamelCase_		     =    1 / 0.18215 * latents
       lowerCamelCase_		     =    self.vae.decode(A_  ).sample

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

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

       if output_type == "pil":
             lowerCamelCase_		     =    self.numpy_to_pil(A_  )

       if not return_dict:
             return image

       return StableDiffusionPipelineOutput(images=A_						,							nsfw_content_detected=A_  )









def 			_SCREAMING_SNAKE_CASE			(							lowercase	:       float	,							lowercase	:       list[float]     ):


      '''simple docstring'''



      if discount_rate < 0:
            raise ValueError('Discount rate cannot be negative'     )
      if not cash_flows:
            raise ValueError('Cash flows list cannot be empty'     )
      lowerCamelCase_		     =    sum(
          cash_flow / ((1 + discount_rate) ** i) for i, cash_flow in enumerate(lowercase     )     )
      return round(lowercase	,							ndigits=2     )


if __name__ == "__main__":
 import doctest

 doctest.testmod()



