






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			:					Union[str, Any]         =	logging.get_logger(__name__)  # pylint: disable=invalid-name



class 							_UpperCamelCase			(a_			):


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


      def  __UpperCAmelCase		(	self					,						__UpperCamelCase = "auto"	)->	List[str]:
           if slice_size == "auto":
                __lowerCAmelCase								=					self.unet.config.attention_head_dim // 2
           self.unet.set_attention_slice(__UpperCamelCase	)


      def  __UpperCAmelCase		(	self	)->	Tuple:
           self.enable_attention_slicing(__UpperCamelCase	)


      @torch.no_grad()
      def __call__(	self					,						__UpperCamelCase					,						__UpperCamelCase=1_6_0_0_0					,						__UpperCamelCase = 5_1_2					,						__UpperCamelCase = 5_1_2					,						__UpperCamelCase = 5_0					,						__UpperCamelCase = 7.5					,						__UpperCamelCase = None					,						__UpperCamelCase = 1					,						__UpperCamelCase = 0.0					,						__UpperCamelCase = None					,						__UpperCamelCase = None					,						__UpperCamelCase = "pil"					,						__UpperCamelCase = True					,						__UpperCamelCase = None					,						__UpperCamelCase = 1					,						**__UpperCamelCase					,						)->	str:
           __lowerCAmelCase								=					self.speech_processor.feature_extractor(
               __UpperCamelCase					,						return_tensors="pt"					,						sampling_rate=__UpperCamelCase	).input_features.to(self.device	)
           __lowerCAmelCase								=					self.speech_model.generate(__UpperCamelCase					,						max_length=4_8_0_0_0_0	)

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

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

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

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

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

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

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

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

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

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

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

           __lowerCAmelCase								=					1 / 0.1_8_2_1_5 * latents
           __lowerCAmelCase								=					self.vae.decode(__UpperCamelCase	).sample

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

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

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

           if not return_dict:
                return image

           return StableDiffusionPipelineOutput(images=__UpperCamelCase					,						nsfw_content_detected=__UpperCamelCase	)














import os
import sys
import unittest


lowerCamelCase			:					int         =	os.path.abspath(os.path.dirname(os.path.dirname(os.path.dirname(__file__))))
sys.path.append(os.path.join(git_repo_path, '''utils'''))

import check_dummies  # noqa: E402
from check_dummies import create_dummy_files, create_dummy_object, find_backend, read_init  # noqa: E402


# Align TRANSFORMERS_PATH in check_dummies with the current path
lowerCamelCase			:					Any         =	os.path.join(git_repo_path, '''src''', '''diffusers''')



class 							_UpperCamelCase			(unittest.TestCase			):


      def  __UpperCAmelCase		(	self	)->	Optional[int]:
           __lowerCAmelCase								=					find_backend("    if not is_torch_available():"	)
           self.assertEqual(__UpperCamelCase					,						"torch"	)

           # backend_with_underscore = find_backend("    if not is_tensorflow_text_available():")
           # self.assertEqual(backend_with_underscore, "tensorflow_text")

           __lowerCAmelCase								=					find_backend("    if not (is_torch_available() and is_transformers_available()):"	)
           self.assertEqual(__UpperCamelCase					,						"torch_and_transformers"	)

           # double_backend_with_underscore = find_backend(
           #    "    if not (is_sentencepiece_available() and is_tensorflow_text_available()):"
           # )
           # self.assertEqual(double_backend_with_underscore, "sentencepiece_and_tensorflow_text")

           __lowerCAmelCase								=					find_backend(
               "    if not (is_torch_available() and is_transformers_available() and is_onnx_available()):"	)
           self.assertEqual(__UpperCamelCase					,						"torch_and_transformers_and_onnx"	)


      def  __UpperCAmelCase		(	self	)->	Dict:
           __lowerCAmelCase								=					read_init()
           # We don't assert on the exact list of keys to allow for smooth grow of backend-specific objects
           self.assertIn("torch"					,						__UpperCamelCase	)
           self.assertIn("torch_and_transformers"					,						__UpperCamelCase	)
           self.assertIn("flax_and_transformers"					,						__UpperCamelCase	)
           self.assertIn("torch_and_transformers_and_onnx"					,						__UpperCamelCase	)

           # Likewise, we can't assert on the exact content of a key
           self.assertIn("UNet2DModel"					,						objects["torch"]	)
           self.assertIn("FlaxUNet2DConditionModel"					,						objects["flax"]	)
           self.assertIn("StableDiffusionPipeline"					,						objects["torch_and_transformers"]	)
           self.assertIn("FlaxStableDiffusionPipeline"					,						objects["flax_and_transformers"]	)
           self.assertIn("LMSDiscreteScheduler"					,						objects["torch_and_scipy"]	)
           self.assertIn("OnnxStableDiffusionPipeline"					,						objects["torch_and_transformers_and_onnx"]	)


      def  __UpperCAmelCase		(	self	)->	Any:
           __lowerCAmelCase								=					create_dummy_object("CONSTANT"					,						"'torch'"	)
           self.assertEqual(__UpperCamelCase					,						"\nCONSTANT = None\n"	)

           __lowerCAmelCase								=					create_dummy_object("function"					,						"'torch'"	)
           self.assertEqual(
               __UpperCamelCase					,						"\ndef function(*args, **kwargs):\n    requires_backends(function, 'torch')\n"	)

           __lowerCAmelCase								=					"\nclass FakeClass(metaclass=DummyObject):\n    _backends = 'torch'\n\n    def __init__(self, *args, **kwargs):\n        requires_backends(self, 'torch')\n\n    @classmethod\n    def from_config(cls, *args, **kwargs):\n        requires_backends(cls, 'torch')\n\n    @classmethod\n    def from_pretrained(cls, *args, **kwargs):\n        requires_backends(cls, 'torch')\n"
           __lowerCAmelCase								=					create_dummy_object("FakeClass"					,						"'torch'"	)
           self.assertEqual(__UpperCamelCase					,						__UpperCamelCase	)


      def  __UpperCAmelCase		(	self	)->	Tuple:
           __lowerCAmelCase								=					"# This file is autogenerated by the command `make fix-copies`, do not edit.\nfrom ..utils import DummyObject, requires_backends\n\n\nCONSTANT = None\n\n\ndef function(*args, **kwargs):\n    requires_backends(function, [\"torch\"])\n\n\nclass FakeClass(metaclass=DummyObject):\n    _backends = [\"torch\"]\n\n    def __init__(self, *args, **kwargs):\n        requires_backends(self, [\"torch\"])\n\n    @classmethod\n    def from_config(cls, *args, **kwargs):\n        requires_backends(cls, [\"torch\"])\n\n    @classmethod\n    def from_pretrained(cls, *args, **kwargs):\n        requires_backends(cls, [\"torch\"])\n"
           __lowerCAmelCase								=					create_dummy_files({"torch": ["CONSTANT", "function", "FakeClass"]}	)
           self.assertEqual(dummy_files["torch"]					,						__UpperCamelCase	)






