






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  UpperCamelCase_      (			snake_case_				):



	'''simple docstring'''




	def __init__(       self       ,			a       ,			a       ,			a       ,			a       ,			a       ,			a       ,			a       ,			a       ,			a       ,			)       ->    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=a       ,			speech_processor=a       ,			vae=a       ,			text_encoder=a       ,			tokenizer=a       ,			unet=a       ,			scheduler=a       ,			feature_extractor=a       ,			)




	def 							_UpperCamelCase			(       self       ,			a = "auto"					)       ->    Any:
								if slice_size == "auto":
															snake_case_				=							self.unet.config.attention_head_dim // 2
								self.unet.set_attention_slice(a					)




	def 							_UpperCamelCase			(       self					)       ->    Tuple:
								self.enable_attention_slicing(a					)






	@torch.no_grad()
	def __call__(       self       ,			a       ,			a=1_60_00       ,			a = 5_12       ,			a = 5_12       ,			a = 50       ,			a = 7.5       ,			a = None       ,			a = 1       ,			a = 0.0       ,			a = None       ,			a = None       ,			a = "pil"       ,			a = True       ,			a = None       ,			a = 1       ,			**a       ,			)       ->    Union[str, Any]:
								snake_case_				=							self.speech_processor.feature_extractor(
								    a       ,			return_tensors='pt'       ,			sampling_rate=a					).input_features.to(self.device					)
								snake_case_				=							self.speech_model.generate(a       ,			max_length=48_00_00					)

								snake_case_				=							self.speech_processor.tokenizer.batch_decode(a       ,			skip_special_tokens=a       ,			normalize=a					)[
								    0
								]

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

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

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

															snake_case_				=							text_input_ids.shape[-1]
															snake_case_				=							self.tokenizer(
															    a       ,			padding='max_length'       ,			max_length=a       ,			truncation=a       ,			return_tensors='pt'       ,			)
															snake_case_				=							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_				=							uncond_embeddings.shape[1]
															snake_case_				=							uncond_embeddings.repeat(1       ,			a       ,			1					)
															snake_case_				=							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
															snake_case_				=							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_				=							(batch_size * num_images_per_prompt, self.unet.config.in_channels, height // 8, width // 8)
								snake_case_				=							text_embeddings.dtype
								if latents is None:
															if self.device.type == "mps":
																						# randn does not exist on mps
																						snake_case_				=							torch.randn(a       ,			generator=a       ,			device='cpu'       ,			dtype=a					).to(
																						    self.device					)
															else:
																						snake_case_				=							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}'''					)
															snake_case_				=							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
								snake_case_				=							self.scheduler.timesteps.to(self.device					)

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

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

															# predict the noise residual
															snake_case_				=							self.unet(a       ,			a       ,			encoder_hidden_states=a					).sample

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

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

								snake_case_				=							1 / 0.18_215 * latents
								snake_case_				=							self.vae.decode(a					).sample

								snake_case_				=							(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_				=							image.cpu().permute(0       ,			2       ,			3       ,			1					).float().numpy()

								if output_type == "pil":
															snake_case_				=							self.numpy_to_pil(a					)

								if not return_dict:
															return image

								return StableDiffusionPipelineOutput(images=a       ,			nsfw_content_detected=a					)













from typing import TYPE_CHECKING

from ...utils import _LazyModule


lowercase  							=					{"processing_wav2vec2_with_lm": ["Wav2Vec2ProcessorWithLM"]}


if TYPE_CHECKING:
	from .processing_wavaveca_with_lm import WavaVecaProcessorWithLM
else:
	import sys

	lowercase  							=					_LazyModule(__name__, globals()["__file__"], _import_structure, module_spec=__spec__)





