

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 				__UpperCamelCase				(					SCREAMING_SNAKE_CASE			):







	def __init__(   self: Optional[Any]  , __UpperCamelCase: WhisperForConditionalGeneration  , __UpperCamelCase: WhisperProcessor  , __UpperCamelCase: AutoencoderKL  , __UpperCamelCase: CLIPTextModel  , __UpperCamelCase: CLIPTokenizer  , __UpperCamelCase: UNetaDConditionModel  , __UpperCamelCase: Union[DDIMScheduler, PNDMScheduler, LMSDiscreteScheduler]  , __UpperCamelCase: StableDiffusionSafetyChecker  , __UpperCamelCase: CLIPImageProcessor  , ):

						'''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=__UpperCamelCase  , speech_processor=__UpperCamelCase  , vae=__UpperCamelCase  , text_encoder=__UpperCamelCase  , tokenizer=__UpperCamelCase  , unet=__UpperCamelCase  , scheduler=__UpperCamelCase  , feature_extractor=__UpperCamelCase  , )







	def 	_SCREAMING_SNAKE_CASE							(   self: Any  , __UpperCamelCase: Optional[Union[str, int]] = "auto"      ):

						'''simple docstring'''





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







	def 	_SCREAMING_SNAKE_CASE							(   self: Union[str, Any]      ):

						'''simple docstring'''





						self.enable_attention_slicing(__UpperCamelCase      )







	@torch.no_grad()
	def __call__(   self: List[str]  , __UpperCamelCase: List[str]  , __UpperCamelCase: int=1_60_00  , __UpperCamelCase: int = 5_12  , __UpperCamelCase: int = 5_12  , __UpperCamelCase: int = 50  , __UpperCamelCase: float = 7.5  , __UpperCamelCase: Optional[Union[str, List[str]]] = None  , __UpperCamelCase: Optional[int] = 1  , __UpperCamelCase: float = 0.0  , __UpperCamelCase: Optional[torch.Generator] = None  , __UpperCamelCase: Optional[torch.FloatTensor] = None  , __UpperCamelCase: Optional[str] = "pil"  , __UpperCamelCase: bool = True  , __UpperCamelCase: Optional[Callable[[int, int, torch.FloatTensor], None]] = None  , __UpperCamelCase: int = 1  , **__UpperCamelCase: Optional[Any]  , ):

						'''simple docstring'''





						__magic_name__							=    self.speech_processor.feature_extractor(
						    __UpperCamelCase  , return_tensors='pt'  , sampling_rate=__UpperCamelCase      ).input_features.to(self.device      )
						__magic_name__							=    self.speech_model.generate(__UpperCamelCase  , max_length=48_00_00      )

						__magic_name__							=    self.speech_processor.tokenizer.batch_decode(__UpperCamelCase  , skip_special_tokens=__UpperCamelCase  , normalize=__UpperCamelCase      )[
						    0
						]

						if isinstance(__UpperCamelCase  , __UpperCamelCase      ):
											__magic_name__							=    1
						elif isinstance(__UpperCamelCase  , __UpperCamelCase      ):
											__magic_name__							=    len(__UpperCamelCase      )
						else:
											raise ValueError(F'`prompt` has to be of type `str` or `list` but is {type(__UpperCamelCase      )}'      )

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

						# get prompt text embeddings
						__magic_name__							=    self.tokenizer(
						    __UpperCamelCase  , padding='max_length'  , max_length=self.tokenizer.model_max_length  , return_tensors='pt'  , )
						__magic_name__							=    text_inputs.input_ids

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

						# duplicate text embeddings for each generation per prompt, using mps friendly method
						__magic_name__,	__magic_name__,	__magic_name__							=    text_embeddings.shape
						__magic_name__							=    text_embeddings.repeat(1  , __UpperCamelCase  , 1      )
						__magic_name__							=    text_embeddings.view(bs_embed * num_images_per_prompt  , __UpperCamelCase  , -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.
						__magic_name__							=    guidance_scale > 1.0
						# get unconditional embeddings for classifier free guidance
						if do_classifier_free_guidance:
											__magic_name__							=    42
											if negative_prompt is None:
																__magic_name__							=    [''] * batch_size
											elif type(__UpperCamelCase      ) is not type(__UpperCamelCase      ):
																raise TypeError(
																    F'`negative_prompt` should be the same type to `prompt`, but got {type(__UpperCamelCase      )} !='
																    F' {type(__UpperCamelCase      )}.'      )
											elif isinstance(__UpperCamelCase  , __UpperCamelCase      ):
																__magic_name__							=    [negative_prompt]
											elif batch_size != len(__UpperCamelCase      ):
																raise ValueError(
																    F'`negative_prompt`: {negative_prompt} has batch size {len(__UpperCamelCase      )}, but `prompt`:'
																    F' {prompt} has batch size {batch_size}. Please make sure that passed `negative_prompt` matches'
																    ' the batch size of `prompt`.'      )
											else:
																__magic_name__							=    negative_prompt

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

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

						# set timesteps
						self.scheduler.set_timesteps(__UpperCamelCase      )

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

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

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

											# predict the noise residual
											__magic_name__							=    self.unet(__UpperCamelCase  , __UpperCamelCase  , encoder_hidden_states=__UpperCamelCase      ).sample

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

											# compute the previous noisy sample x_t -> x_t-1
											__magic_name__							=    self.scheduler.step(__UpperCamelCase  , __UpperCamelCase  , __UpperCamelCase  , **__UpperCamelCase      ).prev_sample

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

						__magic_name__							=    1 / 0.18215 * latents
						__magic_name__							=    self.vae.decode(__UpperCamelCase      ).sample

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

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

						if output_type == "pil":
											__magic_name__							=    self.numpy_to_pil(__UpperCamelCase      )

						if not return_dict:
											return image

						return StableDiffusionPipelineOutput(images=__UpperCamelCase  , nsfw_content_detected=__UpperCamelCase      )







import shutil
import tempfile
import unittest

from transformers import ClapFeatureExtractor, ClapProcessor, RobertaTokenizer, RobertaTokenizerFast
from transformers.testing_utils import require_sentencepiece, require_torchaudio

from .test_feature_extraction_clap import floats_list

@require_torchaudio
@require_sentencepiece
class 				__UpperCamelCase				(					unittest.TestCase			):







	def 	_SCREAMING_SNAKE_CASE							(   self: Optional[Any]      ):

						'''simple docstring'''





						__magic_name__							=    'laion/clap-htsat-unfused'
						__magic_name__							=    tempfile.mkdtemp()







	def 	_SCREAMING_SNAKE_CASE							(   self: Union[str, Any]  , **__UpperCamelCase: List[Any]      ):

						'''simple docstring'''





						return RobertaTokenizer.from_pretrained(self.checkpoint  , **__UpperCamelCase      )







	def 	_SCREAMING_SNAKE_CASE							(   self: int  , **__UpperCamelCase: Dict      ):

						'''simple docstring'''





						return ClapFeatureExtractor.from_pretrained(self.checkpoint  , **__UpperCamelCase      )







	def 	_SCREAMING_SNAKE_CASE							(   self: Optional[int]      ):

						'''simple docstring'''





						shutil.rmtree(self.tmpdirname      )







	def 	_SCREAMING_SNAKE_CASE							(   self: Optional[Any]      ):

						'''simple docstring'''





						__magic_name__							=    self.get_tokenizer()
						__magic_name__							=    self.get_feature_extractor()

						__magic_name__							=    ClapProcessor(tokenizer=__UpperCamelCase  , feature_extractor=__UpperCamelCase      )

						processor.save_pretrained(self.tmpdirname      )
						__magic_name__							=    ClapProcessor.from_pretrained(self.tmpdirname      )

						self.assertEqual(processor.tokenizer.get_vocab()  , tokenizer.get_vocab()      )
						self.assertIsInstance(processor.tokenizer  , __UpperCamelCase      )

						self.assertEqual(processor.feature_extractor.to_json_string()  , feature_extractor.to_json_string()      )
						self.assertIsInstance(processor.feature_extractor  , __UpperCamelCase      )







	def 	_SCREAMING_SNAKE_CASE							(   self: Union[str, Any]      ):

						'''simple docstring'''





						__magic_name__							=    ClapProcessor(tokenizer=self.get_tokenizer()  , feature_extractor=self.get_feature_extractor()      )
						processor.save_pretrained(self.tmpdirname      )

						__magic_name__							=    self.get_tokenizer(bos_token='(BOS)'  , eos_token='(EOS)'      )
						__magic_name__							=    self.get_feature_extractor(do_normalize=__UpperCamelCase  , padding_value=1.0      )

						__magic_name__							=    ClapProcessor.from_pretrained(
						    self.tmpdirname  , bos_token='(BOS)'  , eos_token='(EOS)'  , do_normalize=__UpperCamelCase  , padding_value=1.0      )

						self.assertEqual(processor.tokenizer.get_vocab()  , tokenizer_add_kwargs.get_vocab()      )
						self.assertIsInstance(processor.tokenizer  , __UpperCamelCase      )

						self.assertEqual(processor.feature_extractor.to_json_string()  , feature_extractor_add_kwargs.to_json_string()      )
						self.assertIsInstance(processor.feature_extractor  , __UpperCamelCase      )







	def 	_SCREAMING_SNAKE_CASE							(   self: List[str]      ):

						'''simple docstring'''





						__magic_name__							=    self.get_feature_extractor()
						__magic_name__							=    self.get_tokenizer()

						__magic_name__							=    ClapProcessor(tokenizer=__UpperCamelCase  , feature_extractor=__UpperCamelCase      )

						__magic_name__							=    floats_list((3, 10_00)      )

						__magic_name__							=    feature_extractor(__UpperCamelCase  , return_tensors='np'      )
						__magic_name__							=    processor(audios=__UpperCamelCase  , return_tensors='np'      )

						for key in input_feat_extract.keys():
											self.assertAlmostEqual(input_feat_extract[key].sum()  , input_processor[key].sum()  , delta=1E-2      )







	def 	_SCREAMING_SNAKE_CASE							(   self: List[Any]      ):

						'''simple docstring'''





						__magic_name__							=    self.get_feature_extractor()
						__magic_name__							=    self.get_tokenizer()

						__magic_name__							=    ClapProcessor(tokenizer=__UpperCamelCase  , feature_extractor=__UpperCamelCase      )

						__magic_name__							=    'This is a test string'

						__magic_name__							=    processor(text=__UpperCamelCase      )

						__magic_name__							=    tokenizer(__UpperCamelCase      )

						for key in encoded_tok.keys():
											self.assertListEqual(encoded_tok[key]  , encoded_processor[key]      )







	def 	_SCREAMING_SNAKE_CASE							(   self: Optional[Any]      ):

						'''simple docstring'''





						__magic_name__							=    self.get_feature_extractor()
						__magic_name__							=    self.get_tokenizer()

						__magic_name__							=    ClapProcessor(tokenizer=__UpperCamelCase  , feature_extractor=__UpperCamelCase      )

						__magic_name__							=    [[1, 4, 5, 8, 1, 0, 8], [3, 4, 3, 1, 1, 8, 9]]

						__magic_name__							=    processor.batch_decode(__UpperCamelCase      )
						__magic_name__							=    tokenizer.batch_decode(__UpperCamelCase      )

						self.assertListEqual(__UpperCamelCase  , __UpperCamelCase      )







	def 	_SCREAMING_SNAKE_CASE							(   self: Any      ):

						'''simple docstring'''





						__magic_name__							=    self.get_feature_extractor()
						__magic_name__							=    self.get_tokenizer()

						__magic_name__							=    ClapProcessor(tokenizer=__UpperCamelCase  , feature_extractor=__UpperCamelCase      )

						self.assertListEqual(
						    processor.model_input_names[2:]  , feature_extractor.model_input_names  , msg='`processor` and `feature_extractor` model input names do not match'  , )




