

"""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     snake_case	(				_snake_case  ):

				def __init__(	self							:				List[str]     ,  UpperCamelCase__							:				str     ,  UpperCamelCase__							:				Dict     ,  UpperCamelCase__							:				Optional[int]     ,  UpperCamelCase__							:				Union[str, Any]     ,  UpperCamelCase__							:				Tuple     ,  UpperCamelCase__							:				int     ,  UpperCamelCase__							:				Optional[int]     ,  UpperCamelCase__							:				Optional[Any]     ,  UpperCamelCase__							:				List[str]     ,  )->   int:



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

				def 							lowercase_     (	self							:				List[str]     ,  UpperCamelCase__							:				Tuple = "auto")->   List[Any]:



						'''simple docstring'''


						if slice_size == "auto":
								__lowerCAmelCase:       List[str]          =			self.unet.config.attention_head_dim // 2
						self.unet.set_attention_slice(_lowerCamelCase)

				def 							lowercase_     (	self							:				int)->   Any:



						'''simple docstring'''


						self.enable_attention_slicing(_lowerCamelCase)




				@torch.no_grad()
				def __call__(	self							:				List[str]     ,  UpperCamelCase__							:				List[Any]     ,  UpperCamelCase__							:				Optional[int]=1_6_0_0_0     ,  UpperCamelCase__							:				List[Any] = 5_1_2     ,  UpperCamelCase__							:				Optional[int] = 5_1_2     ,  UpperCamelCase__							:				Tuple = 5_0     ,  UpperCamelCase__							:				Any = 7.5     ,  UpperCamelCase__							:				Optional[int] = None     ,  UpperCamelCase__							:				int = 1     ,  UpperCamelCase__							:				Union[str, Any] = 0.0     ,  UpperCamelCase__							:				Any = None     ,  UpperCamelCase__							:				Any = None     ,  UpperCamelCase__							:				str = "pil"     ,  UpperCamelCase__							:				Tuple = True     ,  UpperCamelCase__							:				int = None     ,  UpperCamelCase__							:				str = 1     ,  **UpperCamelCase__							:				Union[str, Any]     ,  )->   Dict:



						'''simple docstring'''


						__lowerCAmelCase:       Union[str, Any]          =			self.speech_processor.feature_extractor(
						    _lowerCamelCase     ,  return_tensors="pt"     ,  sampling_rate=_lowerCamelCase).input_features.to(self.device)
						__lowerCAmelCase:       Optional[Any]          =			self.speech_model.generate(_lowerCamelCase     ,  max_length=4_8_0_0_0_0)

						__lowerCAmelCase:       List[Any]          =			self.speech_processor.tokenizer.batch_decode(_lowerCamelCase     ,  skip_special_tokens=_lowerCamelCase     ,  normalize=_lowerCamelCase)[
						    0
						]

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

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

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

								__lowerCAmelCase:       List[Any]          =			text_input_ids.shape[-1]
								__lowerCAmelCase:       Optional[Any]          =			self.tokenizer(
								    _lowerCamelCase     ,  padding="max_length"     ,  max_length=_lowerCamelCase     ,  truncation=_lowerCamelCase     ,  return_tensors="pt"     ,  )
								__lowerCAmelCase:       Optional[int]          =			self.text_encoder(uncond_input.input_ids.to(self.device))[0]

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

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

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

								# predict the noise residual
								__lowerCAmelCase:       List[str]          =			self.unet(_lowerCamelCase     ,  _lowerCamelCase     ,  encoder_hidden_states=_lowerCamelCase).sample

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

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

						__lowerCAmelCase:       Any          =			1 / 0.18215 * latents
						__lowerCAmelCase:       Tuple          =			self.vae.decode(_lowerCamelCase).sample

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

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

						if not return_dict:
								return image

						return StableDiffusionPipelineOutput(images=_lowerCamelCase     ,  nsfw_content_detected=_lowerCamelCase)






from typing import List, Optional, Union

import torch

from ...models import UNetaDConditionModel, VQModel
from ...pipelines import DiffusionPipeline
from ...pipelines.pipeline_utils import ImagePipelineOutput
from ...schedulers import DDPMScheduler
from ...utils import (
    is_accelerate_available,
    is_accelerate_version,
    logging,
    randn_tensor,
    replace_example_docstring,
)


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

snake_case				:    Union[str, Any]       				=     '''
    Examples:
        ```py
        >>> from diffusers import KandinskyV22Pipeline, KandinskyV22PriorPipeline
        >>> import torch

        >>> pipe_prior = KandinskyV22PriorPipeline.from_pretrained("kandinsky-community/kandinsky-2-2-prior")
        >>> pipe_prior.to("cuda")
        >>> prompt = "red cat, 4k photo"
        >>> out = pipe_prior(prompt)
        >>> image_emb = out.image_embeds
        >>> zero_image_emb = out.negative_image_embeds
        >>> pipe = KandinskyV22Pipeline.from_pretrained("kandinsky-community/kandinsky-2-2-decoder")
        >>> pipe.to("cuda")
        >>> image = pipe(
        ...     image_embeds=image_emb,
        ...     negative_image_embeds=zero_image_emb,
        ...     height=768,
        ...     width=768,
        ...     num_inference_steps=50,
        ... ).images
        >>> image[0].save("cat.png")
        ```
'''
def 				__lowerCamelCase						( UpperCAmelCase_						:		List[Any] ,  UpperCAmelCase_						:		Optional[int] ,  UpperCAmelCase_						:		List[str]=8						):




      """simple docstring"""





      a						:List[str] 							=       height // scale_factor**2
      if height % scale_factor**2 != 0:
            new_height += 1
      a						:int 							=       width // scale_factor**2
      if width % scale_factor**2 != 0:
            new_width += 1
      return new_height * scale_factor, new_width * scale_factor







class 		_snake_case  (      _snake_case							):




 def __init__(		self					,  _lowerCamelCase					,  _lowerCamelCase					,  _lowerCamelCase					,  ):
       super().__init__()

       self.register_modules(
           unet=_lowerCamelCase					,  scheduler=_lowerCamelCase					,  movq=_lowerCamelCase					,  )
       a						:Any 							=       2 ** (len(self.movq.config.block_out_channels    ) - 1)




 def        SCREAMING_SNAKE_CASE__				(		self					,  _lowerCamelCase					,  _lowerCamelCase					,  _lowerCamelCase					,  _lowerCamelCase					,  _lowerCamelCase					,  _lowerCamelCase    ):
       if latents is None:
             a						:str 							=       randn_tensor(_lowerCamelCase					,  generator=_lowerCamelCase					,  device=_lowerCamelCase					,  dtype=_lowerCamelCase    )
       else:
             if latents.shape != shape:
                   raise ValueError(F'''Unexpected latents shape, got {latents.shape}, expected {shape}'''    )
             a						:Any 							=       latents.to(_lowerCamelCase    )

       a						:Dict 							=       latents * scheduler.init_noise_sigma
       return latents




 def        SCREAMING_SNAKE_CASE__				(		self					,  _lowerCamelCase=0    ):
       if is_accelerate_available():
             from accelerate import cpu_offload
       else:
             raise ImportError('''Please install accelerate via `pip install accelerate`'''    )

       a						:int 							=       torch.device(F'''cuda:{gpu_id}'''    )

       a						:int 							=       [
           self.unet,
           self.movq,
       ]
       for cpu_offloaded_model in models:
             if cpu_offloaded_model is not None:
                   cpu_offload(_lowerCamelCase					,  _lowerCamelCase    )




 def        SCREAMING_SNAKE_CASE__				(		self					,  _lowerCamelCase=0    ):
       if is_accelerate_available() and is_accelerate_version('''>='''					,  '''0.17.0.dev0'''    ):
             from accelerate import cpu_offload_with_hook
       else:
             raise ImportError('''`enable_model_cpu_offload` requires `accelerate v0.17.0` or higher.'''    )

       a						:Any 							=       torch.device(F'''cuda:{gpu_id}'''    )

       if self.device.type != "cpu":
             self.to('''cpu'''					,  silence_dtype_warnings=_lowerCamelCase    )
             torch.cuda.empty_cache()  # otherwise we don't see the memory savings (but they probably exist)

       a						:Tuple 							=       None
       for cpu_offloaded_model in [self.unet, self.movq]:
             a							,				a						:List[str] 							=       cpu_offload_with_hook(_lowerCamelCase					,  _lowerCamelCase					,  prev_module_hook=_lowerCamelCase    )

       # We'll offload the last model manually.
       a						:str 							=       hook




 @property
 # Copied from diffusers.pipelines.stable_diffusion.pipeline_stable_diffusion.StableDiffusionPipeline._execution_device
 def        SCREAMING_SNAKE_CASE__				(		self    ):
       if not hasattr(self.unet					,  '''_hf_hook'''    ):
             return self.device
       for module in self.unet.modules():
             if (
                 hasattr(_lowerCamelCase					,  '''_hf_hook'''    )
                 and hasattr(module._hf_hook					,  '''execution_device'''    )
                 and module._hf_hook.execution_device is not None
             ):
                   return torch.device(module._hf_hook.execution_device    )
       return self.device




 @torch.no_grad()
 @replace_example_docstring(_lowerCamelCase    )
 def __call__(		self					,  _lowerCamelCase					,  _lowerCamelCase					,  _lowerCamelCase = 512					,  _lowerCamelCase = 512					,  _lowerCamelCase = 100					,  _lowerCamelCase = 4.0					,  _lowerCamelCase = 1					,  _lowerCamelCase = None					,  _lowerCamelCase = None					,  _lowerCamelCase = "pil"					,  _lowerCamelCase = True					,  ):
       a						:int 							=       self._execution_device

       a						:Optional[Any] 							=       guidance_scale > 1.0

       if isinstance(_lowerCamelCase					,  _lowerCamelCase    ):
             a						:Union[str, Any] 							=       torch.cat(_lowerCamelCase					,  dim=0    )
       a						:Any 							=       image_embeds.shape[0] * num_images_per_prompt
       if isinstance(_lowerCamelCase					,  _lowerCamelCase    ):
             a						:List[str] 							=       torch.cat(_lowerCamelCase					,  dim=0    )

       if do_classifier_free_guidance:
             a						:Union[str, Any] 							=       image_embeds.repeat_interleave(_lowerCamelCase					,  dim=0    )
             a						:Optional[int] 							=       negative_image_embeds.repeat_interleave(_lowerCamelCase					,  dim=0    )

       a						:Optional[int] 							=       torch.cat([negative_image_embeds, image_embeds]					,  dim=0    ).to(dtype=self.unet.dtype					,  device=_lowerCamelCase    )

       self.scheduler.set_timesteps(_lowerCamelCase					,  device=_lowerCamelCase    )
       a						:Optional[Any] 							=       self.scheduler.timesteps

       a						:List[str] 							=       self.unet.config.in_channels

       a							,				a						:str 							=       downscale_height_and_width(_lowerCamelCase					,  _lowerCamelCase					,  self.movq_scale_factor    )

       # create initial latent
       a						:int 							=       self.prepare_latents(
           (batch_size, num_channels_latents, height, width)					,  image_embeds.dtype					,  _lowerCamelCase					,  _lowerCamelCase					,  _lowerCamelCase					,  self.scheduler					,  )

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

             a						:Union[str, Any] 							=       {'''image_embeds''': image_embeds}
             a						:Optional[Any] 							=       self.unet(
                 sample=_lowerCamelCase					,  timestep=_lowerCamelCase					,  encoder_hidden_states=_lowerCamelCase					,  added_cond_kwargs=_lowerCamelCase					,  return_dict=_lowerCamelCase					,  )[0]

             if do_classifier_free_guidance:
                   a							,				a						:Any 							=       noise_pred.split(latents.shape[1]					,  dim=1    )
                   a							,				a						:List[str] 							=       noise_pred.chunk(2    )
                   a							,				a						:int 							=       variance_pred.chunk(2    )
                   a						:List[Any] 							=       noise_pred_uncond + guidance_scale * (noise_pred_text - noise_pred_uncond)
                   a						:Optional[int] 							=       torch.cat([noise_pred, variance_pred_text]					,  dim=1    )

             if not (
                 hasattr(self.scheduler.config					,  '''variance_type'''    )
                 and self.scheduler.config.variance_type in ["learned", "learned_range"]
             ):
                   a							,				a						:Tuple 							=       noise_pred.split(latents.shape[1]					,  dim=1    )

             # compute the previous noisy sample x_t -> x_t-1
             a						:int 							=       self.scheduler.step(
                 _lowerCamelCase					,  _lowerCamelCase					,  _lowerCamelCase					,  generator=_lowerCamelCase					,  )[0]
       # post-processing
       a						:int 							=       self.movq.decode(_lowerCamelCase					,  force_not_quantize=_lowerCamelCase    )['''sample''']

       if output_type not in ["pt", "np", "pil"]:
             raise ValueError(F'''Only the output types `pt`, `pil` and `np` are supported not output_type={output_type}'''    )

       if output_type in ["np", "pil"]:
             a						:str 							=       image * 0.5 + 0.5
             a						:List[Any] 							=       image.clamp(0					,  1    )
             a						:str 							=       image.cpu().permute(0					,  2					,  3					,  1    ).float().numpy()

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

       if not return_dict:
             return (image,)

       return ImagePipelineOutput(images=_lowerCamelCase    )





