






"""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_       (__A     ):





	def __init__(      self					: int	,	lowerCAmelCase_					: WhisperForConditionalGeneration	,	lowerCAmelCase_					: WhisperProcessor	,	lowerCAmelCase_					: AutoencoderKL	,	lowerCAmelCase_					: CLIPTextModel	,	lowerCAmelCase_					: CLIPTokenizer	,	lowerCAmelCase_					: UNetaDConditionModel	,	lowerCAmelCase_					: Union[DDIMScheduler, PNDMScheduler, LMSDiscreteScheduler]	,	lowerCAmelCase_					: StableDiffusionSafetyChecker	,	lowerCAmelCase_					: CLIPImageProcessor	,	)						->			Optional[Any]:
				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					: List[Any]	,	lowerCAmelCase_					: Optional[Union[str, int]] = "auto"						)						->			Dict:
				if slice_size == "auto":
							UpperCAmelCase_     : Dict			     =					self.unet.config.attention_head_dim // 2
				self.unet.set_attention_slice(lowerCAmelCase_						)





	def      _SCREAMING_SNAKE_CASE      (      self					: List[Any]						)						->			Optional[int]:
				self.enable_attention_slicing(lowerCAmelCase_						)





	@torch.no_grad()
	def __call__(      self					: Any	,	lowerCAmelCase_					: Any	,	lowerCAmelCase_					: List[Any]=16_000	,	lowerCAmelCase_					: int = 512	,	lowerCAmelCase_					: int = 512	,	lowerCAmelCase_					: int = 50	,	lowerCAmelCase_					: float = 7.5	,	lowerCAmelCase_					: Optional[Union[str, List[str]]] = None	,	lowerCAmelCase_					: Optional[int] = 1	,	lowerCAmelCase_					: float = 0.0	,	lowerCAmelCase_					: Optional[torch.Generator] = None	,	lowerCAmelCase_					: Optional[torch.FloatTensor] = None	,	lowerCAmelCase_					: Optional[str] = "pil"	,	lowerCAmelCase_					: bool = True	,	lowerCAmelCase_					: Optional[Callable[[int, int, torch.FloatTensor], None]] = None	,	lowerCAmelCase_					: int = 1	,	**lowerCAmelCase_					: Union[str, Any]	,	)						->			List[str]:
				UpperCAmelCase_     : List[str]			     =					self.speech_processor.feature_extractor(
				    lowerCAmelCase_	,	return_tensors="pt"	,	sampling_rate=lowerCAmelCase_						).input_features.to(self.device						)
				UpperCAmelCase_     : int			     =					self.speech_model.generate(lowerCAmelCase_	,	max_length=480_000						)

				UpperCAmelCase_     : Any			     =					self.speech_processor.tokenizer.batch_decode(lowerCAmelCase_	,	skip_special_tokens=lowerCAmelCase_	,	normalize=lowerCAmelCase_						)[
				    0
				]

				if isinstance(lowerCAmelCase_	,	lowerCAmelCase_						):
							UpperCAmelCase_     : List[str]			     =					1
				elif isinstance(lowerCAmelCase_	,	lowerCAmelCase_						):
							UpperCAmelCase_     : Optional[int]			     =					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_     : Optional[int]			     =					self.tokenizer(
				    lowerCAmelCase_	,	padding="max_length"	,	max_length=self.tokenizer.model_max_length	,	return_tensors="pt"	,	)
				UpperCAmelCase_     : List[str]			     =					text_inputs.input_ids

				if text_input_ids.shape[-1] > self.tokenizer.model_max_length:
							UpperCAmelCase_     : Dict			     =					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_     : List[str]			     =					text_input_ids[:, : self.tokenizer.model_max_length]
				UpperCAmelCase_     : Dict			     =					self.text_encoder(text_input_ids.to(self.device						)						)[0]

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

							UpperCAmelCase_     : int			     =					text_input_ids.shape[-1]
							UpperCAmelCase_     : str			     =					self.tokenizer(
							    lowerCAmelCase_	,	padding="max_length"	,	max_length=lowerCAmelCase_	,	truncation=lowerCAmelCase_	,	return_tensors="pt"	,	)
							UpperCAmelCase_     : Tuple			     =					self.text_encoder(uncond_input.input_ids.to(self.device						)						)[0]

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

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

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

							# predict the noise residual
							UpperCAmelCase_     : Any			     =					self.unet(lowerCAmelCase_	,	lowerCAmelCase_	,	encoder_hidden_states=lowerCAmelCase_						).sample

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

							# compute the previous noisy sample x_t -> x_t-1
							UpperCAmelCase_     : Tuple			     =					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_     : List[str]			     =					1 / 0.1_8_2_1_5 * latents
				UpperCAmelCase_     : Tuple			     =					self.vae.decode(lowerCAmelCase_						).sample

				UpperCAmelCase_     : List[Any]			     =					(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_     : Optional[Any]			     =					image.cpu().permute(0	,	2	,	3	,	1						).float().numpy()

				if output_type == "pil":
							UpperCAmelCase_     : Union[str, Any]			     =					self.numpy_to_pil(lowerCAmelCase_						)

				if not return_dict:
							return image

				return StableDiffusionPipelineOutput(images=lowerCAmelCase_	,	nsfw_content_detected=lowerCAmelCase_						)











"""simple docstring"""
import warnings

from diffusers import StableDiffusionInpaintPipeline as StableDiffusionInpaintPipeline  # noqa F401


warnings.warn(
    '''The `inpainting.py` script is outdated. Please use directly `from diffusers import'''
    ''' StableDiffusionInpaintPipeline` instead.'''
)



