



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


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




class       lowercase__      (   __UpperCAmelCase     ):
							'''simple docstring'''







							def __init__(	self						,							snake_case						,							snake_case						,							snake_case						,							snake_case						,							snake_case						,							snake_case						,							snake_case						,							snake_case						,							snake_case						,							)      ->				Tuple:
									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 							lowerCamelCase_ (	self						,							snake_case = "auto"    )      ->				Optional[Any]:
									if slice_size == "auto":
											_UpperCAmelCase							=		self.unet.config.attention_head_dim // 2
									self.unet.set_attention_slice(lowerCAmelCase_    )






							def 							lowerCamelCase_ (	self    )      ->				int:
									self.enable_attention_slicing(lowerCAmelCase_    )






							@torch.no_grad()
							def __call__(	self						,							snake_case						,							snake_case=16000						,							snake_case = 512						,							snake_case = 512						,							snake_case = 50						,							snake_case = 7.5						,							snake_case = None						,							snake_case = 1						,							snake_case = 0.0						,							snake_case = None						,							snake_case = None						,							snake_case = "pil"						,							snake_case = True						,							snake_case = None						,							snake_case = 1						,							**snake_case						,							)      ->				Dict:
									_UpperCAmelCase							=		self.speech_processor.feature_extractor(
									    lowerCAmelCase_						,							return_tensors='pt'						,							sampling_rate=lowerCAmelCase_    ).input_features.to(self.device    )
									_UpperCAmelCase							=		self.speech_model.generate(lowerCAmelCase_						,							max_length=480000    )

									_UpperCAmelCase							=		self.speech_processor.tokenizer.batch_decode(lowerCAmelCase_						,							skip_special_tokens=lowerCAmelCase_						,							normalize=lowerCAmelCase_    )[
									    0
									]

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

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

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

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

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

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

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

											# predict the noise residual
											_UpperCAmelCase							=		self.unet(lowerCAmelCase_						,							lowerCAmelCase_						,							encoder_hidden_states=lowerCAmelCase_    ).sample

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

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

									_UpperCAmelCase							=		1 / 0.18215 * latents
									_UpperCAmelCase							=		self.vae.decode(lowerCAmelCase_    ).sample

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

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

									if output_type == "pil":
											_UpperCAmelCase							=		self.numpy_to_pil(lowerCAmelCase_    )

									if not return_dict:
											return image

									return StableDiffusionPipelineOutput(images=lowerCAmelCase_						,							nsfw_content_detected=lowerCAmelCase_    )










"""simple docstring"""
def 				lowercase__		(					lowerCamelCase,						lowerCamelCase       ):
				return numa ^ numa < 0


if __name__ == "__main__":
			import doctest

			doctest.testmod()

