





'''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


_SCREAMING_SNAKE_CASE							     =			logging.get_logger(__name__)  # pylint: disable=invalid-name
class 			_lowerCAmelCase		(			__A							,     __A	):






     """simple docstring"""






     @register_to_config
     def __init__(   self							:			str   ,       __snake_case							:			int   ,       __snake_case							:			List[Any] = None   ,       __snake_case							:			int = None						)->	str:
        super().__init__()

        snake_case 						=   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"

           snake_case 						=   torch.zeros(snake_case_   ,       snake_case_						)
        else:
           snake_case 						=   None

        snake_case 						=   torch.nn.Parameter(snake_case_						)




class 			_lowerCAmelCase		(			__A	):






     """simple docstring"""






     snake_case_						      =   42
     snake_case_						      =   42
     snake_case_						      =   42
     snake_case_						      =   42
     snake_case_						      =   42
     snake_case_						      =   42



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

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



     def      lowerCAmelCase			(   self							:			str   ,       __snake_case							:			List[str]   ,       __snake_case							:			str   ,       __snake_case							:			Optional[Any]						)->	Tuple:
        snake_case 						=   len(snake_case_						) if isinstance(snake_case_   ,       snake_case_						) else 1

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

        if text_input_ids.shape[-1] > self.tokenizer.model_max_length:
           snake_case 						=   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}'''						)
           snake_case 						=   text_input_ids[:, : self.tokenizer.model_max_length]
        snake_case 						=   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
        snake_case 						=   prompt_embeds / prompt_embeds.norm(dim=-1   ,       keepdim=snake_case_						)

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

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

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

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

        return prompt_embeds



     @torch.no_grad()
     def __call__(   self							:			Optional[Any]   ,       __snake_case							:			Tuple   ,       __snake_case							:			Optional[Any] = 1_00   ,       __snake_case							:			Dict = 5.0   ,       __snake_case							:			Optional[Any] = 1.0   ,       __snake_case							:			Tuple = 1   ,       __snake_case							:			Optional[Any] = None   ,       __snake_case							:			Union[str, Any] = None   ,       __snake_case							:			Optional[int] = "pil"   ,       __snake_case							:			Optional[int] = True   ,       __snake_case							:			Dict = None   ,       __snake_case							:			Tuple = 1   ,       )->	Optional[int]:


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

        snake_case 						=   batch_size * num_images_per_prompt

        snake_case 						=   guidance_scale > 1.0

        snake_case 						=   self._encode_prompt(snake_case_   ,       snake_case_   ,       snake_case_						)

        if (callback_steps is None) or (
            callback_steps is not None and (not isinstance(snake_case_   ,       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(snake_case_						)}.'''						)

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

        snake_case 						=   (batch_size, self.transformer.num_latent_pixels)
        if latents is None:
           snake_case 						=   self.transformer.num_vector_embeds - 1
           snake_case 						=   torch.full(snake_case_   ,       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).'''						)
           snake_case 						=   latents.to(self.device						)

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

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

        snake_case 						=   latents

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

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

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

           snake_case 						=   self.truncate(snake_case_   ,       snake_case_						)

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

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

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

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

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

        if output_type == "pil":
           snake_case 						=   self.numpy_to_pil(snake_case_						)

        if not return_dict:
           return (image,)

        return ImagePipelineOutput(images=snake_case_						)



     def      lowerCAmelCase			(   self							:			List[str]   ,       __snake_case							:			Union[str, Any]   ,       __snake_case							:			Any						)->	int:
        snake_case 						=   torch.sort(snake_case_   ,       1   ,       descending=snake_case_						)
        snake_case 						=   torch.exp(snake_case_						)
        snake_case 						=   sorted_p_x_0.cumsum(dim=1						) < truncation_rate

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

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

        snake_case 						=   log_p_x_0.clone()

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

        return rv


"""simple docstring"""



import copy
from typing import Any, Dict, List, Optional, Union

import numpy as np
import torch

from ...audio_utils import mel_filter_bank, spectrogram, window_function
from ...feature_extraction_sequence_utils import SequenceFeatureExtractor
from ...feature_extraction_utils import BatchFeature
from ...utils import TensorType, logging


_UpperCamelCase								=		logging.get_logger(__name__)
class 						__UpperCAmelCase    (__A     ):







			'''simple docstring'''
			_UpperCamelCase  :						Optional[int]					     =				['input_features', 'is_longer']





			def __init__(			self   ,      snake_case_=64   ,      snake_case_=48_000   ,      snake_case_=480   ,      snake_case_=10   ,      snake_case_=1_024   ,      snake_case_=0.0   ,      snake_case_=False   ,      snake_case_ = 0   ,      snake_case_ = 14_000   ,      snake_case_ = None   ,      snake_case_ = "fusion"   ,      snake_case_ = "repeatpad"   ,      **snake_case_   ,      ):







										'''simple docstring'''


										super().__init__(
										    feature_size=snake_case_   ,      sampling_rate=snake_case_   ,      padding_value=snake_case_   ,      return_attention_mask=snake_case_   ,      **snake_case_   ,      )
										A__       :     List[str]								=			top_db
										A__       :     Optional[int]								=			truncation
										A__       :     str								=			padding
										A__       :     str								=			fft_window_size
										A__       :     Any								=			(fft_window_size >> 1) + 1
										A__       :     Optional[Any]								=			hop_length
										A__       :     Any								=			max_length_s
										A__       :     int								=			max_length_s * sampling_rate
										A__       :     Any								=			sampling_rate
										A__       :     List[str]								=			frequency_min
										A__       :     int								=			frequency_max
										A__       :     Tuple								=			mel_filter_bank(
										    num_frequency_bins=self.nb_frequency_bins   ,      num_mel_filters=snake_case_   ,      min_frequency=snake_case_   ,      max_frequency=snake_case_   ,      sampling_rate=snake_case_   ,      norm=snake_case_   ,      mel_scale="""htk"""   ,      )
										A__       :     List[Any]								=			mel_filter_bank(
										    num_frequency_bins=self.nb_frequency_bins   ,      num_mel_filters=snake_case_   ,      min_frequency=snake_case_   ,      max_frequency=snake_case_   ,      sampling_rate=snake_case_   ,      norm="""slaney"""   ,      mel_scale="""slaney"""   ,      )





			def 		lowerCamelCase					(			self							):







										'''simple docstring'''


										A__       :     List[Any]								=			copy.deepcopy(self.__dict__							)
										A__       :     int								=			self.__class__.__name__
										if "mel_filters" in output:
																	del output["mel_filters"]
										if "mel_filters_slaney" in output:
																	del output["mel_filters_slaney"]
										return output





			def 		lowerCamelCase					(			self   ,      snake_case_   ,      snake_case_ = None							):







										'''simple docstring'''


										A__       :     Any								=			spectrogram(
										    snake_case_   ,      window_function(self.fft_window_size   ,      """hann"""							)   ,      frame_length=self.fft_window_size   ,      hop_length=self.hop_length   ,      power=2.0   ,      mel_filters=snake_case_   ,      log_mel="""dB"""   ,      )
										return log_mel_spectrogram.T





			def 		lowerCamelCase					(			self   ,      snake_case_   ,      snake_case_   ,      snake_case_							):







										'''simple docstring'''


										A__       :     Union[str, Any]								=			np.array_split(list(range(0   ,      total_frames - chunk_frames + 1							)							)   ,      3							)
										if len(ranges[1]							) == 0:
																	# if the audio is too short, we just use the first chunk
																	A__       :     Optional[Any]								=			[0]
										if len(ranges[2]							) == 0:
																	# if the audio is too short, we just use the first chunk
																	A__       :     str								=			[0]
										# randomly choose index for each part
										A__       :     int								=			np.random.choice(ranges[0]							)
										A__       :     Dict								=			np.random.choice(ranges[1]							)
										A__       :     Dict								=			np.random.choice(ranges[2]							)

										A__       :     Any								=			mel[idx_front : idx_front + chunk_frames, :]
										A__       :     Optional[int]								=			mel[idx_middle : idx_middle + chunk_frames, :]
										A__       :     Optional[int]								=			mel[idx_back : idx_back + chunk_frames, :]

										A__       :     Dict								=			torch.tensor(mel[None, None, :]							)
										A__       :     Dict								=			torch.nn.functional.interpolate(
										    snake_case_   ,      size=[chunk_frames, 64]   ,      mode="""bilinear"""   ,      align_corners=snake_case_							)
										A__       :     List[Any]								=			mel_shrink[0][0].numpy()
										A__       :     Dict								=			np.stack([mel_shrink, mel_chunk_front, mel_chunk_middle, mel_chunk_back]   ,      axis=0							)
										return mel_fusion





			def 		lowerCamelCase					(			self   ,      snake_case_   ,      snake_case_   ,      snake_case_   ,      snake_case_							):







										'''simple docstring'''


										if waveform.shape[0] > max_length:
																	if truncation == "rand_trunc":
																								A__       :     int								=			True
																								# random crop to max_length (for compatibility) -> this should be handled by self.pad
																								A__       :     Tuple								=			len(snake_case_							) - max_length
																								A__       :     Any								=			np.random.randint(0   ,      overflow + 1							)
																								A__       :     int								=			waveform[idx : idx + max_length]
																								A__       :     List[str]								=			self._np_extract_fbank_features(snake_case_   ,      self.mel_filters_slaney							)[None, :]
																	elif truncation == "fusion":
																								A__       :     Optional[int]								=			self._np_extract_fbank_features(snake_case_   ,      self.mel_filters							)
																								A__       :     Optional[int]								=			max_length // self.hop_length + 1  # the +1 related to how the spectrogram is computed
																								A__       :     int								=			mel.shape[0]
																								if chunk_frames == total_frames:
																															# there is a corner case where the audio length is larger than max_length but smaller than max_length+hop_length.
																															# In this case, we just use the whole audio.
																															A__       :     Union[str, Any]								=			np.stack([mel, mel, mel, mel]   ,      axis=0							)
																															A__       :     Dict								=			False
																								else:
																															A__       :     List[Any]								=			self._random_mel_fusion(snake_case_   ,      snake_case_   ,      snake_case_							)
																															A__       :     Optional[Any]								=			True
																	else:
																								raise NotImplementedError(F'''data_truncating {truncation} not implemented'''							)

										else:
																	A__       :     Tuple								=			False
																	# only use repeat as a new possible value for padding. you repeat the audio before applying the usual max_length padding
																	if waveform.shape[0] < max_length:
																								if padding == "repeat":
																															A__       :     Union[str, Any]								=			int(max_length / len(snake_case_							)							)
																															A__       :     List[Any]								=			np.stack(np.tile(snake_case_   ,      n_repeat + 1							)							)[:max_length]
																								if padding == "repeatpad":
																															A__       :     List[Any]								=			int(max_length / len(snake_case_							)							)
																															A__       :     Tuple								=			np.stack(np.tile(snake_case_   ,      snake_case_							)							)
																								A__       :     Tuple								=			np.pad(snake_case_   ,      (0, max_length - waveform.shape[0])   ,      mode="""constant"""   ,      constant_values=0							)

																	if truncation == "fusion":
																								A__       :     str								=			self._np_extract_fbank_features(snake_case_   ,      self.mel_filters							)
																								A__       :     List[Any]								=			np.stack([input_mel, input_mel, input_mel, input_mel]   ,      axis=0							)
																	else:
																								A__       :     List[Any]								=			self._np_extract_fbank_features(snake_case_   ,      self.mel_filters_slaney							)[None, :]

										return input_mel, longer





			def __call__(			self   ,      snake_case_   ,      snake_case_ = None   ,      snake_case_ = None   ,      snake_case_ = None   ,      snake_case_ = None   ,      snake_case_ = None   ,      **snake_case_   ,      ):







										'''simple docstring'''


										A__       :     Optional[Any]								=			truncation if truncation is not None else self.truncation
										A__       :     Union[str, Any]								=			padding if padding else self.padding

										if sampling_rate is not None:
																	if sampling_rate != self.sampling_rate:
																								raise ValueError(
																								    F'''The model corresponding to this feature extractor: {self.__class__.__name__} was trained using a'''
																								    F''' sampling rate of {self.sampling_rate}. Please make sure that the provided `raw_speech` input'''
																								    F''' was sampled with {self.sampling_rate} and not {sampling_rate}.'''							)
										else:
																	logger.warning(
																	    """It is strongly recommended to pass the `sampling_rate` argument to this function. """
																	    """Failing to do so can result in silent errors that might be hard to debug."""							)

										A__       :     List[Any]								=			isinstance(snake_case_   ,      np.ndarray							) and len(raw_speech.shape							) > 1
										if is_batched_numpy and len(raw_speech.shape							) > 2:
																	raise ValueError(F'''Only mono-channel audio is supported for input to {self}'''							)
										A__       :     int								=			is_batched_numpy or (
										    isinstance(snake_case_   ,      (list, tuple)							) and (isinstance(raw_speech[0]   ,      (np.ndarray, tuple, list)							))
										)

										if is_batched:
																	A__       :     Any								=			[np.asarray(snake_case_   ,      dtype=np.floataa							) for speech in raw_speech]
										elif not is_batched and not isinstance(snake_case_   ,      np.ndarray							):
																	A__       :     Any								=			np.asarray(snake_case_   ,      dtype=np.floataa							)
										elif isinstance(snake_case_   ,      np.ndarray							) and raw_speech.dtype is np.dtype(np.floataa							):
																	A__       :     Optional[int]								=			raw_speech.astype(np.floataa							)

										# always return batch
										if not is_batched:
																	A__       :     Union[str, Any]								=			[np.asarray(snake_case_							)]

										# convert to mel spectrogram, truncate and pad if needed.
										A__       :     Optional[Any]								=			[
										    self._get_input_mel(snake_case_   ,      max_length if max_length else self.nb_max_samples   ,      snake_case_   ,      snake_case_							)
										    for waveform in raw_speech
										]

										A__       :     List[str]								=			[]
										A__       :     Optional[Any]								=			[]
										for mel, longer in padded_inputs:
																	input_mel.append(snake_case_							)
																	is_longer.append(snake_case_							)

										if truncation == "fusion" and sum(snake_case_							) == 0:
																	# if no audio is longer than 10s, then randomly select one audio to be longer
																	A__       :     Optional[Any]								=			np.random.randint(0   ,      len(snake_case_							)							)
																	A__       :     str								=			True

										if isinstance(input_mel[0]   ,      snake_case_							):
																	A__       :     Optional[Any]								=			[np.asarray(snake_case_   ,      dtype=np.floataa							) for feature in input_mel]

										# is_longer is a list of bool
										A__       :     Union[str, Any]								=			[[longer] for longer in is_longer]

										A__       :     str								=			{"""input_features""": input_mel, """is_longer""": is_longer}
										A__       :     Optional[int]								=			BatchFeature(snake_case_							)

										if return_tensors is not None:
																	A__       :     Dict								=			input_features.convert_to_tensors(snake_case_							)

										return input_features





