



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


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

class       _a  ( SCREAMING_SNAKE_CASE__):



			def __init__(			self							:				Union[str, Any]   ,	_lowercase							:				WhisperForConditionalGeneration   ,	_lowercase							:				WhisperProcessor   ,	_lowercase							:				AutoencoderKL   ,	_lowercase							:				CLIPTextModel   ,	_lowercase							:				CLIPTokenizer   ,	_lowercase							:				UNetaDConditionModel   ,	_lowercase							:				Union[DDIMScheduler, PNDMScheduler, LMSDiscreteScheduler]   ,	_lowercase							:				StableDiffusionSafetyChecker   ,	_lowercase							:				CLIPImageProcessor   ,	)							->					Optional[int]:
									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=_lowercase   ,	speech_processor=_lowercase   ,	vae=_lowercase   ,	text_encoder=_lowercase   ,	tokenizer=_lowercase   ,	unet=_lowercase   ,	scheduler=_lowercase   ,	feature_extractor=_lowercase   ,	)



			def 				__lowercase					(			self							:				Optional[int]   ,	_lowercase							:				Optional[Union[str, int]] = "auto"		)							->					List[str]:
									if slice_size == "auto":
															snake_case   :			Any						    = self.unet.config.attention_head_dim // 2
									self.unet.set_attention_slice(_lowercase		)



			def 				__lowercase					(			self							:				Optional[int]		)							->					Optional[int]:
									self.enable_attention_slicing(_lowercase		)



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

									snake_case   :			Optional[int]						    = self.speech_processor.tokenizer.batch_decode(_lowercase   ,	skip_special_tokens=_lowercase   ,	normalize=_lowercase		)[
									    0
									]

									if isinstance(_lowercase   ,	_lowercase		):
															snake_case   :			Optional[int]						    = 1
									elif isinstance(_lowercase   ,	_lowercase		):
															snake_case   :			Tuple						    = len(_lowercase		)
									else:
															raise ValueError(F'''`prompt` has to be of type `str` or `list` but is {type(_lowercase		)}'''		)

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

									# get prompt text embeddings
									snake_case   :			Optional[Any]						    = self.tokenizer(
									    _lowercase   ,	padding="max_length"   ,	max_length=self.tokenizer.model_max_length   ,	return_tensors="pt"   ,	)
									snake_case   :			List[Any]						    = text_inputs.input_ids

									if text_input_ids.shape[-1] > self.tokenizer.model_max_length:
															snake_case   :			Tuple						    = 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   :			str						    = text_input_ids[:, : self.tokenizer.model_max_length]
									snake_case   :			str						    = 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   :			Union[str, Any]						    = text_embeddings.shape
									snake_case   :			List[Any]						    = text_embeddings.repeat(1   ,	_lowercase   ,	1		)
									snake_case   :			str						    = text_embeddings.view(bs_embed * num_images_per_prompt   ,	_lowercase   ,	-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   :			Dict						    = guidance_scale > 1.0
									# get unconditional embeddings for classifier free guidance
									if do_classifier_free_guidance:
															snake_case   :			List[str]
															if negative_prompt is None:
																					snake_case   :			Dict						    = [""] * batch_size
															elif type(_lowercase		) is not type(_lowercase		):
																					raise TypeError(
																					    F'''`negative_prompt` should be the same type to `prompt`, but got {type(_lowercase		)} !='''
																					    F''' {type(_lowercase		)}.'''		)
															elif isinstance(_lowercase   ,	_lowercase		):
																					snake_case   :			List[str]						    = [negative_prompt]
															elif batch_size != len(_lowercase		):
																					raise ValueError(
																					    F'''`negative_prompt`: {negative_prompt} has batch size {len(_lowercase		)}, 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   :			List[str]						    = negative_prompt

															snake_case   :			List[str]						    = text_input_ids.shape[-1]
															snake_case   :			Dict						    = self.tokenizer(
															    _lowercase   ,	padding="max_length"   ,	max_length=_lowercase   ,	truncation=_lowercase   ,	return_tensors="pt"   ,	)
															snake_case   :			Any						    = 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   :			int						    = uncond_embeddings.shape[1]
															snake_case   :			Optional[Any]						    = uncond_embeddings.repeat(1   ,	_lowercase   ,	1		)
															snake_case   :			int						    = uncond_embeddings.view(batch_size * num_images_per_prompt   ,	_lowercase   ,	-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   :			Optional[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`.
									snake_case   :			int						    = (batch_size * num_images_per_prompt, self.unet.config.in_channels, height // 8, width // 8)
									snake_case   :			List[str]						    = text_embeddings.dtype
									if latents is None:
															if self.device.type == "mps":
																					# randn does not exist on mps
																					snake_case   :			Dict						    = torch.randn(_lowercase   ,	generator=_lowercase   ,	device="cpu"   ,	dtype=_lowercase		).to(
																					    self.device		)
															else:
																					snake_case   :			Any						    = torch.randn(_lowercase   ,	generator=_lowercase   ,	device=self.device   ,	dtype=_lowercase		)
									else:
															if latents.shape != latents_shape:
																					raise ValueError(F'''Unexpected latents shape, got {latents.shape}, expected {latents_shape}'''		)
															snake_case   :			List[str]						    = latents.to(self.device		)

									# set timesteps
									self.scheduler.set_timesteps(_lowercase		)

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

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

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

															# predict the noise residual
															snake_case   :			Union[str, Any]						    = self.unet(_lowercase   ,	_lowercase   ,	encoder_hidden_states=_lowercase		).sample

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

															# compute the previous noisy sample x_t -> x_t-1
															snake_case   :			Optional[int]						    = self.scheduler.step(_lowercase   ,	_lowercase   ,	_lowercase   ,	**_lowercase		).prev_sample

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

									snake_case   :			List[str]						    = 1 / 0.18215 * latents
									snake_case   :			Union[str, Any]						    = self.vae.decode(_lowercase		).sample

									snake_case   :			int						    = (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   :			Union[str, Any]						    = image.cpu().permute(0   ,	2   ,	3   ,	1		).float().numpy()

									if output_type == "pil":
															snake_case   :			int						    = self.numpy_to_pil(_lowercase		)

									if not return_dict:
															return image

									return StableDiffusionPipelineOutput(images=_lowercase   ,	nsfw_content_detected=_lowercase		)








"""simple docstring"""







import importlib
import shutil
import threading
import warnings
from typing import List

import fsspec
import fsspec.asyn

from . import compression
from .hffilesystem import HfFileSystem


A		=						importlib.util.find_spec('s3fs') is not None

if _has_safs:
			from .safilesystem import SaFileSystem  # noqa: F401

A		=						[
    compression.BzaFileSystem,
    compression.GzipFileSystem,
    compression.LzaFileSystem,
    compression.XzFileSystem,
    compression.ZstdFileSystem,
]

# Register custom filesystems
for fs_class in COMPRESSION_FILESYSTEMS + [HfFileSystem]:
			if fs_class.protocol in fsspec.registry and fsspec.registry[fs_class.protocol] is not fs_class:
						warnings.warn(F'''A filesystem protocol was already set for {fs_class.protocol} and will be overwritten.''')
			fsspec.register_implementation(fs_class.protocol, fs_class, clobber=True)

def  __SCREAMING_SNAKE_CASE		(			lowerCamelCase_:							str ):
						"""simple docstring"""






						if "://" in dataset_path:
												snake_case   :			List[str]						    = dataset_path.split("://" )[1]
						return dataset_path

def  __SCREAMING_SNAKE_CASE		(			lowerCamelCase_:							fsspec.AbstractFileSystem ):
						"""simple docstring"""






						if fs is not None and fs.protocol != "file":
												return True
						else:
												return False

def  __SCREAMING_SNAKE_CASE		(			lowerCamelCase_:							fsspec.AbstractFileSystem     ,		lowerCamelCase_:							str     ,		lowerCamelCase_:							str ):
						"""simple docstring"""






						snake_case   :			int						    = not is_remote_filesystem(lowerCamelCase_ )
						if is_local:
												# LocalFileSystem.mv does copy + rm, it is more efficient to simply move a local directory
												shutil.move(fs._strip_protocol(lowerCamelCase_ )     ,		fs._strip_protocol(lowerCamelCase_ ) )
						else:
												fs.mv(lowerCamelCase_     ,		lowerCamelCase_     ,		recursive=lowerCamelCase_ )

def  __SCREAMING_SNAKE_CASE		(			):
						"""simple docstring"""






						if hasattr(fsspec.asyn     ,		"reset_lock" ):
												# for future fsspec>2022.05.0
												fsspec.asyn.reset_lock()
						else:
												snake_case   :			Union[str, Any]						    = None
												snake_case   :			Optional[int]						    = None
												snake_case   :			List[str]						    = threading.Lock()



