






'''simple docstring'''




from typing import Callable, List, Optional, Tuple, Union

import torch
from transformers import CLIPTextModel, CLIPTokenizer

from ...configuration_utils import ConfigMixin, register_to_config
from ...models import ModelMixin, TransformeraDModel, VQModel
from ...schedulers import VQDiffusionScheduler
from ...utils import logging
from ..pipeline_utils import DiffusionPipeline, ImagePipelineOutput


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

class    _UpperCAmelCase			(     snake_case_							,    snake_case_						):




   """simple docstring"""

   @register_to_config
   def __init__(		self :	Union[str, Any]      ,      __UpperCAmelCase :	bool      ,      __UpperCAmelCase :	Optional[int] = None      ,      __UpperCAmelCase :	Optional[int] = None		):





         '''simple docstring'''





         super().__init__()

         _A			     =	learnable

         if self.learnable:
               assert hidden_size is not None, "learnable=True requires `hidden_size` to be set"
               assert length is not None, "learnable=True requires `length` to be set"

               _A			     =	torch.zeros(__UpperCAmelCase      ,      __UpperCAmelCase		)
         else:
               _A			     =	None

         _A			     =	torch.nn.Parameter(__UpperCAmelCase		)





class    _UpperCAmelCase			(     snake_case_						):




   """simple docstring"""

   snake_case    				=						42
   snake_case    				=						42
   snake_case    				=						42
   snake_case    				=						42
   snake_case    				=						42
   snake_case    				=						42

   def __init__(		self :	Any      ,      __UpperCAmelCase :	VQModel      ,      __UpperCAmelCase :	CLIPTextModel      ,      __UpperCAmelCase :	CLIPTokenizer      ,      __UpperCAmelCase :	TransformeraDModel      ,      __UpperCAmelCase :	VQDiffusionScheduler      ,      __UpperCAmelCase :	LearnedClassifierFreeSamplingEmbeddings      ,      ):





         '''simple docstring'''





         super().__init__()

         self.register_modules(
             vqvae=__UpperCAmelCase      ,      transformer=__UpperCAmelCase      ,      text_encoder=__UpperCAmelCase      ,      tokenizer=__UpperCAmelCase      ,      scheduler=__UpperCAmelCase      ,      learned_classifier_free_sampling_embeddings=__UpperCAmelCase      ,      )

   def 						lowerCAmelCase (		self :	int      ,      __UpperCAmelCase :	Union[str, Any]      ,      __UpperCAmelCase :	Dict      ,      __UpperCAmelCase :	Any		):





         '''simple docstring'''





         _A			     =	len(__UpperCAmelCase		) if isinstance(__UpperCAmelCase      ,      __UpperCAmelCase		) else 1

         # get prompt text embeddings
         _A			     =	self.tokenizer(
             __UpperCAmelCase      ,      padding="max_length"      ,      max_length=self.tokenizer.model_max_length      ,      return_tensors="pt"      ,      )
         _A			     =	text_inputs.input_ids

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

         # NOTE: This additional step of normalizing the text embeddings is from VQ-Diffusion.
         # While CLIP does normalize the pooled output of the text transformer when combining
         # the image and text embeddings, CLIP does not directly normalize the last hidden state.
         #
         # CLIP normalizing the pooled output.
         # https://github.com/huggingface/transformers/blob/d92e22d1f28324f513f3080e5c47c071a3916721/src/transformers/models/clip/modeling_clip.py#L1052-L1053
         _A			     =	prompt_embeds / prompt_embeds.norm(dim=-1      ,      keepdim=__UpperCAmelCase		)

         # duplicate text embeddings for each generation per prompt
         _A			     =	prompt_embeds.repeat_interleave(__UpperCAmelCase      ,      dim=0		)

         if do_classifier_free_guidance:
               if self.learned_classifier_free_sampling_embeddings.learnable:
                     _A			     =	self.learned_classifier_free_sampling_embeddings.embeddings
                     _A			     =	negative_prompt_embeds.unsqueeze(0		).repeat(__UpperCAmelCase      ,      1      ,      1		)
               else:
                     _A			     =	[""] * batch_size

                     _A			     =	text_input_ids.shape[-1]
                     _A			     =	self.tokenizer(
                         __UpperCAmelCase      ,      padding="max_length"      ,      max_length=__UpperCAmelCase      ,      truncation=__UpperCAmelCase      ,      return_tensors="pt"      ,      )
                     _A			     =	self.text_encoder(uncond_input.input_ids.to(self.device		)		)[0]
                     # See comment for normalizing text embeddings
                     _A			     =	negative_prompt_embeds / negative_prompt_embeds.norm(dim=-1      ,      keepdim=__UpperCAmelCase		)

               # duplicate unconditional embeddings for each generation per prompt, using mps friendly method
               _A			     =	negative_prompt_embeds.shape[1]
               _A			     =	negative_prompt_embeds.repeat(1      ,      __UpperCAmelCase      ,      1		)
               _A			     =	negative_prompt_embeds.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
               _A			     =	torch.cat([negative_prompt_embeds, prompt_embeds]		)

         return prompt_embeds

   @torch.no_grad()
   def __call__(		self :	Optional[Any]      ,      __UpperCAmelCase :	Union[str, List[str]]      ,      __UpperCAmelCase :	int = 100      ,      __UpperCAmelCase :	float = 5.0      ,      __UpperCAmelCase :	float = 1.0      ,      __UpperCAmelCase :	int = 1      ,      __UpperCAmelCase :	Optional[Union[torch.Generator, List[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      ,      ):





         '''simple docstring'''





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

         _A			     =	batch_size * num_images_per_prompt

         _A			     =	guidance_scale > 1.0

         _A			     =	self._encode_prompt(__UpperCAmelCase      ,      __UpperCAmelCase      ,      __UpperCAmelCase		)

         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 the initial completely masked latents unless the user supplied it

         _A			     =	(batch_size, self.transformer.num_latent_pixels)
         if latents is None:
               _A			     =	self.transformer.num_vector_embeds - 1
               _A			     =	torch.full(__UpperCAmelCase      ,      __UpperCAmelCase		).to(self.device		)
         else:
               if latents.shape != latents_shape:
                     raise ValueError(f'''Unexpected latents shape, got {latents.shape}, expected {latents_shape}'''		)
               if (latents < 0).any() or (latents >= self.transformer.num_vector_embeds).any():
                     raise ValueError(
                         "Unexpected latents value(s). All latents be valid embedding indices i.e. in the range 0,"
                         f''' {self.transformer.num_vector_embeds - 1} (inclusive).'''		)
               _A			     =	latents.to(self.device		)

         # set timesteps
         self.scheduler.set_timesteps(__UpperCAmelCase      ,      device=self.device		)

         _A			     =	self.scheduler.timesteps.to(self.device		)

         _A			     =	latents

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

               # predict the un-noised image
               # model_output == `log_p_x_0`
               _A			     =	self.transformer(__UpperCAmelCase      ,      encoder_hidden_states=__UpperCAmelCase      ,      timestep=__UpperCAmelCase		).sample

               if do_classifier_free_guidance:
                     _A     ,	_A			     =	model_output.chunk(2		)
                     _A			     =	model_output_uncond + guidance_scale * (model_output_text - model_output_uncond)
                     model_output -= torch.logsumexp(__UpperCAmelCase      ,      dim=1      ,      keepdim=__UpperCAmelCase		)

               _A			     =	self.truncate(__UpperCAmelCase      ,      __UpperCAmelCase		)

               # remove `log(0)`'s (`-inf`s)
               _A			     =	model_output.clamp(-70		)

               # compute the previous noisy sample x_t -> x_t-1
               _A			     =	self.scheduler.step(__UpperCAmelCase      ,      timestep=__UpperCAmelCase      ,      sample=__UpperCAmelCase      ,      generator=__UpperCAmelCase		).prev_sample

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

         _A			     =	self.vqvae.config.vq_embed_dim
         _A			     =	(batch_size, self.transformer.height, self.transformer.width, embedding_channels)
         _A			     =	self.vqvae.quantize.get_codebook_entry(__UpperCAmelCase      ,      shape=__UpperCAmelCase		)
         _A			     =	self.vqvae.decode(__UpperCAmelCase      ,      force_not_quantize=__UpperCAmelCase		).sample

         _A			     =	(image / 2 + 0.5).clamp(0      ,      1		)
         _A			     =	image.cpu().permute(0      ,      2      ,      3      ,      1		).numpy()

         if output_type == "pil":
               _A			     =	self.numpy_to_pil(__UpperCAmelCase		)

         if not return_dict:
               return (image,)

         return ImagePipelineOutput(images=__UpperCAmelCase		)

   def 						lowerCAmelCase (		self :	Union[str, Any]      ,      __UpperCAmelCase :	torch.FloatTensor      ,      __UpperCAmelCase :	float		):





         '''simple docstring'''





         _A     ,	_A			     =	torch.sort(__UpperCAmelCase      ,      1      ,      descending=__UpperCAmelCase		)
         _A			     =	torch.exp(__UpperCAmelCase		)
         _A			     =	sorted_p_x_0.cumsum(dim=1		) < truncation_rate

         # Ensure that at least the largest probability is not zeroed out
         _A			     =	torch.full_like(keep_mask[:, 0:1, :]      ,      __UpperCAmelCase		)
         _A			     =	torch.cat((all_true, keep_mask)      ,      dim=1		)
         _A			     =	keep_mask[:, :-1, :]

         _A			     =	keep_mask.gather(1      ,      indices.argsort(1		)		)

         _A			     =	log_p_x_0.clone()

         _A			     =	-torch.inf  # -inf = log(0)

         return rv









from ...processing_utils import ProcessorMixin
from ...tokenization_utils_base import BatchEncoding







class 		_snake_case  (      _snake_case							):
 SCREAMING_SNAKE_CASE__      =       'ClapFeatureExtractor'
 SCREAMING_SNAKE_CASE__      =       ('RobertaTokenizer', 'RobertaTokenizerFast')




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




 def __call__(		self					,  _lowerCamelCase=None					,  _lowerCamelCase=None					,  _lowerCamelCase=None					,  **_lowerCamelCase    ):
       a						:Dict 							=       kwargs.pop('''sampling_rate'''					,  _lowerCamelCase    )

       if text is None and audios is None:
             raise ValueError('''You have to specify either text or audios. Both cannot be none.'''    )

       if text is not None:
             a						:Optional[int] 							=       self.tokenizer(_lowerCamelCase					,  return_tensors=_lowerCamelCase					,  **_lowerCamelCase    )

       if audios is not None:
             a						:Tuple 							=       self.feature_extractor(
                 _lowerCamelCase					,  sampling_rate=_lowerCamelCase					,  return_tensors=_lowerCamelCase					,  **_lowerCamelCase    )

       if text is not None and audios is not None:
             a						:Union[str, Any] 							=       audio_features.input_features
             return encoding
       elif text is not None:
             return encoding
       else:
             return BatchEncoding(data=dict(**_lowerCamelCase    )					,  tensor_type=_lowerCamelCase    )




 def        SCREAMING_SNAKE_CASE__				(		self					,  *_lowerCamelCase					,  **_lowerCamelCase    ):
       return self.tokenizer.batch_decode(*_lowerCamelCase					,  **_lowerCamelCase    )




 def        SCREAMING_SNAKE_CASE__				(		self					,  *_lowerCamelCase					,  **_lowerCamelCase    ):
       return self.tokenizer.decode(*_lowerCamelCase					,  **_lowerCamelCase    )




 @property
 def        SCREAMING_SNAKE_CASE__				(		self    ):
       a						:List[Any] 							=       self.tokenizer.model_input_names
       a						:str 							=       self.feature_extractor.model_input_names
       return list(dict.fromkeys(tokenizer_input_names + feature_extractor_input_names    )    )





