


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


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



class        a					(			_UpperCAmelCase     ,_UpperCAmelCase   ):





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

        __lowerCamelCase:				str     =	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"

           __lowerCamelCase:				Dict     =	torch.zeros(SCREAMING_SNAKE_CASE_    ,	SCREAMING_SNAKE_CASE_							)
        else:
           __lowerCamelCase:				Optional[Any]     =	None

        __lowerCamelCase:				List[Any]     =	torch.nn.Parameter(SCREAMING_SNAKE_CASE_							)



class        a					(			_UpperCAmelCase   ):
     UpperCAmelCase__ :  VQModel
     UpperCAmelCase__ :  CLIPTextModel
     UpperCAmelCase__ :  CLIPTokenizer
     UpperCAmelCase__ :  TransformeraDModel
     UpperCAmelCase__ :  LearnedClassifierFreeSamplingEmbeddings
     UpperCAmelCase__ :  VQDiffusionScheduler





     def __init__(    self					:				int    ,	SCREAMING_SNAKE_CASE_					:				VQModel    ,	SCREAMING_SNAKE_CASE_					:				CLIPTextModel    ,	SCREAMING_SNAKE_CASE_					:				CLIPTokenizer    ,	SCREAMING_SNAKE_CASE_					:				TransformeraDModel    ,	SCREAMING_SNAKE_CASE_					:				VQDiffusionScheduler    ,	SCREAMING_SNAKE_CASE_					:				LearnedClassifierFreeSamplingEmbeddings    ,	):
        super().__init__()

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





     def  SCREAMING_SNAKE_CASE__       (    self					:				Union[str, Any]    ,	SCREAMING_SNAKE_CASE_					:				Optional[Any]    ,	SCREAMING_SNAKE_CASE_					:				int    ,	SCREAMING_SNAKE_CASE_					:				Optional[int]							):
        __lowerCamelCase:				Dict     =	len(SCREAMING_SNAKE_CASE_							) if isinstance(SCREAMING_SNAKE_CASE_    ,	SCREAMING_SNAKE_CASE_							) else 1

        # get prompt text embeddings
        __lowerCamelCase:				str     =	self.tokenizer(
            SCREAMING_SNAKE_CASE_    ,	padding="""max_length"""    ,	max_length=self.tokenizer.model_max_length    ,	return_tensors="""pt"""    ,	)
        __lowerCamelCase:				List[str]     =	text_inputs.input_ids

        if text_input_ids.shape[-1] > self.tokenizer.model_max_length:
           __lowerCamelCase:				int     =	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:				int     =	text_input_ids[:, : self.tokenizer.model_max_length]
        __lowerCamelCase:				Any     =	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
        __lowerCamelCase:				str     =	prompt_embeds / prompt_embeds.norm(dim=-1    ,	keepdim=SCREAMING_SNAKE_CASE_							)

        # duplicate text embeddings for each generation per prompt
        __lowerCamelCase:				Union[str, Any]     =	prompt_embeds.repeat_interleave(SCREAMING_SNAKE_CASE_    ,	dim=0							)

        if do_classifier_free_guidance:
           if self.learned_classifier_free_sampling_embeddings.learnable:
              __lowerCamelCase:				Dict     =	self.learned_classifier_free_sampling_embeddings.embeddings
              __lowerCamelCase:				str     =	negative_prompt_embeds.unsqueeze(0							).repeat(SCREAMING_SNAKE_CASE_    ,	1    ,	1							)
           else:
              __lowerCamelCase:				Union[str, Any]     =	[""""""] * batch_size

              __lowerCamelCase:				int     =	text_input_ids.shape[-1]
              __lowerCamelCase:				Any     =	self.tokenizer(
                  SCREAMING_SNAKE_CASE_    ,	padding="""max_length"""    ,	max_length=SCREAMING_SNAKE_CASE_    ,	truncation=SCREAMING_SNAKE_CASE_    ,	return_tensors="""pt"""    ,	)
              __lowerCamelCase:				Optional[int]     =	self.text_encoder(uncond_input.input_ids.to(self.device							)							)[0]
              # See comment for normalizing text embeddings
              __lowerCamelCase:				Union[str, Any]     =	negative_prompt_embeds / negative_prompt_embeds.norm(dim=-1    ,	keepdim=SCREAMING_SNAKE_CASE_							)

           # duplicate unconditional embeddings for each generation per prompt, using mps friendly method
           __lowerCamelCase:				Tuple     =	negative_prompt_embeds.shape[1]
           __lowerCamelCase:				Optional[Any]     =	negative_prompt_embeds.repeat(1    ,	SCREAMING_SNAKE_CASE_    ,	1							)
           __lowerCamelCase:				List[str]     =	negative_prompt_embeds.view(batch_size * num_images_per_prompt    ,	SCREAMING_SNAKE_CASE_    ,	-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:				List[str]     =	torch.cat([negative_prompt_embeds, prompt_embeds]							)

        return prompt_embeds





     @torch.no_grad()
     def __call__(    self					:				Optional[Any]    ,	SCREAMING_SNAKE_CASE_					:				Union[str, List[str]]    ,	SCREAMING_SNAKE_CASE_					:				int = 100    ,	SCREAMING_SNAKE_CASE_					:				float = 5.0    ,	SCREAMING_SNAKE_CASE_					:				float = 1.0    ,	SCREAMING_SNAKE_CASE_					:				int = 1    ,	SCREAMING_SNAKE_CASE_					:				Optional[Union[torch.Generator, List[torch.Generator]]] = None    ,	SCREAMING_SNAKE_CASE_					:				Optional[torch.FloatTensor] = None    ,	SCREAMING_SNAKE_CASE_					:				Optional[str] = "pil"    ,	SCREAMING_SNAKE_CASE_					:				bool = True    ,	SCREAMING_SNAKE_CASE_					:				Optional[Callable[[int, int, torch.FloatTensor], None]] = None    ,	SCREAMING_SNAKE_CASE_					:				int = 1    ,	):
        if isinstance(SCREAMING_SNAKE_CASE_    ,	SCREAMING_SNAKE_CASE_							):
           __lowerCamelCase:				Optional[Any]     =	1
        elif isinstance(SCREAMING_SNAKE_CASE_    ,	SCREAMING_SNAKE_CASE_							):
           __lowerCamelCase:				Dict     =	len(SCREAMING_SNAKE_CASE_							)
        else:
           raise ValueError(F'`prompt` has to be of type `str` or `list` but is {type(SCREAMING_SNAKE_CASE_							)}'							)

        __lowerCamelCase:				List[Any]     =	batch_size * num_images_per_prompt

        __lowerCamelCase:				int     =	guidance_scale > 1.0

        __lowerCamelCase:				str     =	self._encode_prompt(SCREAMING_SNAKE_CASE_    ,	SCREAMING_SNAKE_CASE_    ,	SCREAMING_SNAKE_CASE_							)

        if (callback_steps is None) or (
            callback_steps is not None and (not isinstance(SCREAMING_SNAKE_CASE_    ,	SCREAMING_SNAKE_CASE_							) or callback_steps <= 0)
        ):
           raise ValueError(
               F'`callback_steps` has to be a positive integer but is {callback_steps} of type'
               F' {type(SCREAMING_SNAKE_CASE_							)}.'							)

        # get the initial completely masked latents unless the user supplied it

        __lowerCamelCase:				Dict     =	(batch_size, self.transformer.num_latent_pixels)
        if latents is None:
           __lowerCamelCase:				Optional[Any]     =	self.transformer.num_vector_embeds - 1
           __lowerCamelCase:				str     =	torch.full(SCREAMING_SNAKE_CASE_    ,	SCREAMING_SNAKE_CASE_							).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).'							)
           __lowerCamelCase:				Union[str, Any]     =	latents.to(self.device							)

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

        __lowerCamelCase:				int     =	self.scheduler.timesteps.to(self.device							)

        __lowerCamelCase:				Union[str, Any]     =	latents

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

           # predict the un-noised image
           # model_output == `log_p_x_0`
           __lowerCamelCase:				str     =	self.transformer(SCREAMING_SNAKE_CASE_    ,	encoder_hidden_states=SCREAMING_SNAKE_CASE_    ,	timestep=SCREAMING_SNAKE_CASE_							).sample

           if do_classifier_free_guidance:
              __lowerCamelCase   ,     __lowerCamelCase:				Any     =	model_output.chunk(2							)
              __lowerCamelCase:				int     =	model_output_uncond + guidance_scale * (model_output_text - model_output_uncond)
              model_output -= torch.logsumexp(SCREAMING_SNAKE_CASE_    ,	dim=1    ,	keepdim=SCREAMING_SNAKE_CASE_							)

           __lowerCamelCase:				str     =	self.truncate(SCREAMING_SNAKE_CASE_    ,	SCREAMING_SNAKE_CASE_							)

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

           # compute the previous noisy sample x_t -> x_t-1
           __lowerCamelCase:				List[Any]     =	self.scheduler.step(SCREAMING_SNAKE_CASE_    ,	timestep=SCREAMING_SNAKE_CASE_    ,	sample=SCREAMING_SNAKE_CASE_    ,	generator=SCREAMING_SNAKE_CASE_							).prev_sample

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

        __lowerCamelCase:				int     =	self.vqvae.config.vq_embed_dim
        __lowerCamelCase:				Union[str, Any]     =	(batch_size, self.transformer.height, self.transformer.width, embedding_channels)
        __lowerCamelCase:				Optional[Any]     =	self.vqvae.quantize.get_codebook_entry(SCREAMING_SNAKE_CASE_    ,	shape=SCREAMING_SNAKE_CASE_							)
        __lowerCamelCase:				List[str]     =	self.vqvae.decode(SCREAMING_SNAKE_CASE_    ,	force_not_quantize=SCREAMING_SNAKE_CASE_							).sample

        __lowerCamelCase:				List[str]     =	(image / 2 + 0.5).clamp(0    ,	1							)
        __lowerCamelCase:				str     =	image.cpu().permute(0    ,	2    ,	3    ,	1							).numpy()

        if output_type == "pil":
           __lowerCamelCase:				Optional[Any]     =	self.numpy_to_pil(SCREAMING_SNAKE_CASE_							)

        if not return_dict:
           return (image,)

        return ImagePipelineOutput(images=SCREAMING_SNAKE_CASE_							)







     def  SCREAMING_SNAKE_CASE__       (    self					:				List[str]    ,	SCREAMING_SNAKE_CASE_					:				torch.FloatTensor    ,	SCREAMING_SNAKE_CASE_					:				float							):
        __lowerCamelCase   ,     __lowerCamelCase:				List[Any]     =	torch.sort(SCREAMING_SNAKE_CASE_    ,	1    ,	descending=SCREAMING_SNAKE_CASE_							)
        __lowerCamelCase:				Optional[Any]     =	torch.exp(SCREAMING_SNAKE_CASE_							)
        __lowerCamelCase:				List[Any]     =	sorted_p_x_0.cumsum(dim=1							) < truncation_rate

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

        __lowerCamelCase:				Optional[Any]     =	keep_mask.gather(1    ,	indices.argsort(1							)							)

        __lowerCamelCase:				Any     =	log_p_x_0.clone()

        __lowerCamelCase:				Tuple     =	-torch.inf  # -inf = log(0)

        return rv






import warnings
from contextlib import contextmanager

from ...processing_utils import ProcessorMixin



class        a					(			_UpperCAmelCase   ):
     UpperCAmelCase__ :  Dict						      =	"Speech2TextFeatureExtractor"
     UpperCAmelCase__ :  str						      =	"Speech2TextTokenizer"





     def __init__(    self					:				Optional[Any]    ,	SCREAMING_SNAKE_CASE_					:				List[str]    ,	SCREAMING_SNAKE_CASE_					:				Optional[int]							):
        super().__init__(SCREAMING_SNAKE_CASE_    ,	SCREAMING_SNAKE_CASE_							)
        __lowerCamelCase:				Any     =	self.feature_extractor
        __lowerCamelCase:				List[Any]     =	False





     def __call__(    self					:				List[str]    ,	*SCREAMING_SNAKE_CASE_					:				Union[str, Any]    ,	**SCREAMING_SNAKE_CASE_					:				List[Any]							):
        # For backward compatibility
        if self._in_target_context_manager:
           return self.current_processor(*SCREAMING_SNAKE_CASE_    ,	**SCREAMING_SNAKE_CASE_							)

        if "raw_speech" in kwargs:
           warnings.warn("""Using `raw_speech` as a keyword argument is deprecated. Use `audio` instead."""							)
           __lowerCamelCase:				Dict     =	kwargs.pop("""raw_speech"""							)
        else:
           __lowerCamelCase:				int     =	kwargs.pop("""audio"""    ,	SCREAMING_SNAKE_CASE_							)
        __lowerCamelCase:				Dict     =	kwargs.pop("""sampling_rate"""    ,	SCREAMING_SNAKE_CASE_							)
        __lowerCamelCase:				int     =	kwargs.pop("""text"""    ,	SCREAMING_SNAKE_CASE_							)
        if len(SCREAMING_SNAKE_CASE_							) > 0:
           __lowerCamelCase:				Dict     =	args[0]
           __lowerCamelCase:				Optional[Any]     =	args[1:]

        if audio is None and text is None:
           raise ValueError("""You need to specify either an `audio` or `text` input to process."""							)

        if audio is not None:
           __lowerCamelCase:				List[str]     =	self.feature_extractor(SCREAMING_SNAKE_CASE_    ,	*SCREAMING_SNAKE_CASE_    ,	sampling_rate=SCREAMING_SNAKE_CASE_    ,	**SCREAMING_SNAKE_CASE_							)
        if text is not None:
           __lowerCamelCase:				Any     =	self.tokenizer(SCREAMING_SNAKE_CASE_    ,	**SCREAMING_SNAKE_CASE_							)

        if text is None:
           return inputs
        elif audio is None:
           return encodings
        else:
           __lowerCamelCase:				List[Any]     =	encodings["""input_ids"""]
           return inputs





     def  SCREAMING_SNAKE_CASE__       (    self					:				List[str]    ,	*SCREAMING_SNAKE_CASE_					:				Optional[Any]    ,	**SCREAMING_SNAKE_CASE_					:				str							):
        return self.tokenizer.batch_decode(*SCREAMING_SNAKE_CASE_    ,	**SCREAMING_SNAKE_CASE_							)





     def  SCREAMING_SNAKE_CASE__       (    self					:				Tuple    ,	*SCREAMING_SNAKE_CASE_					:				Any    ,	**SCREAMING_SNAKE_CASE_					:				List[Any]							):
        return self.tokenizer.decode(*SCREAMING_SNAKE_CASE_    ,	**SCREAMING_SNAKE_CASE_							)







     @contextmanager
     def  SCREAMING_SNAKE_CASE__       (    self					:				Any							):
        warnings.warn(
            """`as_target_processor` is deprecated and will be removed in v5 of Transformers. You can process your """
            """labels by using the argument `text` of the regular `__call__` method (either in the same call as """
            """your audio inputs, or in a separate call."""							)
        __lowerCamelCase:				Any     =	True
        __lowerCamelCase:				Any     =	self.tokenizer
        yield
        __lowerCamelCase:				Any     =	self.feature_extractor
        __lowerCamelCase:				Optional[Any]     =	False


