"""simple docstring"""



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

import numpy as np
import torch

from ...models import TaFilmDecoder
from ...schedulers import DDPMScheduler
from ...utils import is_onnx_available, logging, randn_tensor


if is_onnx_available():
				from ..onnx_utils import OnnxRuntimeModel

from ..pipeline_utils import AudioPipelineOutput, DiffusionPipeline
from .continous_encoder import SpectrogramContEncoder
from .notes_encoder import SpectrogramNotesEncoder


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

_UpperCamelCase								=		2_56
class 						__UpperCAmelCase    (_snake_case     ):







			'''simple docstring'''
			_UpperCamelCase  :						List[Any]					     =				['melgan']





			def __init__(			self   ,      snake_case_   ,      snake_case_   ,      snake_case_   ,      snake_case_   ,      snake_case_   ,      ):







										'''simple docstring'''


										super().__init__()

										# From MELGAN
										A__       :     Optional[int]								=			math.log(1E-5							)  # Matches MelGAN training.
										A__       :     List[Any]								=			4.0  # Largest value for most examples
										A__       :     List[Any]								=			128

										self.register_modules(
										    notes_encoder=__lowerCamelCase   ,      continuous_encoder=__lowerCamelCase   ,      decoder=__lowerCamelCase   ,      scheduler=__lowerCamelCase   ,      melgan=__lowerCamelCase   ,      )





			def 		lowerCamelCase					(			self   ,      snake_case_   ,      snake_case_=(-1.0, 1.0)   ,      snake_case_=False							):







										'''simple docstring'''


										A__       :     Optional[Any]								=			output_range
										if clip:
																	A__       :     Any								=			torch.clip(__lowerCamelCase   ,      self.min_value   ,      self.max_value							)
										# Scale to [0, 1].
										A__       :     Tuple								=			(features - self.min_value) / (self.max_value - self.min_value)
										# Scale to [min_out, max_out].
										return zero_one * (max_out - min_out) + min_out





			def 		lowerCamelCase					(			self   ,      snake_case_   ,      snake_case_=(-1.0, 1.0)   ,      snake_case_=False							):







										'''simple docstring'''


										A__       :     Optional[Any]								=			input_range
										A__       :     Optional[Any]								=			torch.clip(__lowerCamelCase   ,      __lowerCamelCase   ,      __lowerCamelCase							) if clip else outputs
										# Scale to [0, 1].
										A__       :     int								=			(outputs - min_out) / (max_out - min_out)
										# Scale to [self.min_value, self.max_value].
										return zero_one * (self.max_value - self.min_value) + self.min_value





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







										'''simple docstring'''


										A__       :     List[Any]								=			input_tokens > 0
										A__       :     int								=			self.notes_encoder(
										    encoder_input_tokens=__lowerCamelCase   ,      encoder_inputs_mask=__lowerCamelCase							)

										A__       :     int								=			self.continuous_encoder(
										    encoder_inputs=__lowerCamelCase   ,      encoder_inputs_mask=__lowerCamelCase							)

										return [(tokens_encoded, tokens_mask), (continuous_encoded, continuous_mask)]





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







										'''simple docstring'''


										A__       :     List[Any]								=			noise_time
										if not torch.is_tensor(__lowerCamelCase							):
																	A__       :     Optional[Any]								=			torch.tensor([timesteps]   ,      dtype=torch.long   ,      device=input_tokens.device							)
										elif torch.is_tensor(__lowerCamelCase							) and len(timesteps.shape							) == 0:
																	A__       :     str								=			timesteps[None].to(input_tokens.device							)

										# broadcast to batch dimension in a way that's compatible with ONNX/Core ML
										A__       :     int								=			timesteps * torch.ones(input_tokens.shape[0]   ,      dtype=timesteps.dtype   ,      device=timesteps.device							)

										A__       :     Dict								=			self.decoder(
										    encodings_and_masks=__lowerCamelCase   ,      decoder_input_tokens=__lowerCamelCase   ,      decoder_noise_time=__lowerCamelCase							)
										return logits





			@torch.no_grad()
			def __call__(			self   ,      snake_case_   ,      snake_case_ = None   ,      snake_case_ = 100   ,      snake_case_ = True   ,      snake_case_ = "numpy"   ,      snake_case_ = None   ,      snake_case_ = 1   ,      ):







										'''simple docstring'''


										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							)}.'''							)

										A__       :     Any								=			np.zeros([1, TARGET_FEATURE_LENGTH, self.n_dims]   ,      dtype=np.floataa							)
										A__       :     Optional[int]								=			np.zeros([1, 0, self.n_dims]   ,      np.floataa							)
										A__       :     Dict								=			torch.ones((1, TARGET_FEATURE_LENGTH)   ,      dtype=__lowerCamelCase   ,      device=self.device							)

										for i, encoder_input_tokens in enumerate(__lowerCamelCase							):
																	if i == 0:
																								A__       :     Optional[int]								=			torch.from_numpy(pred_mel[:1].copy()							).to(
																								    device=self.device   ,      dtype=self.decoder.dtype							)
																								# The first chunk has no previous context.
																								A__       :     Union[str, Any]								=			torch.zeros((1, TARGET_FEATURE_LENGTH)   ,      dtype=__lowerCamelCase   ,      device=self.device							)
																	else:
																								# The full song pipeline does not feed in a context feature, so the mask
																								# will be all 0s after the feature converter. Because we know we're
																								# feeding in a full context chunk from the previous prediction, set it
																								# to all 1s.
																								A__       :     Optional[Any]								=			ones

																	A__       :     Dict								=			self.scale_features(
																	    __lowerCamelCase   ,      output_range=[-1.0, 1.0]   ,      clip=__lowerCamelCase							)

																	A__       :     Any								=			self.encode(
																	    input_tokens=torch.IntTensor([encoder_input_tokens]							).to(device=self.device							)   ,      continuous_inputs=__lowerCamelCase   ,      continuous_mask=__lowerCamelCase   ,      )

																	# Sample encoder_continuous_inputs shaped gaussian noise to begin loop
																	A__       :     Union[str, Any]								=			randn_tensor(
																	    shape=encoder_continuous_inputs.shape   ,      generator=__lowerCamelCase   ,      device=self.device   ,      dtype=self.decoder.dtype   ,      )

																	# set step values
																	self.scheduler.set_timesteps(__lowerCamelCase							)

																	# Denoising diffusion loop
																	for j, t in enumerate(self.progress_bar(self.scheduler.timesteps							)							):
																								A__       :     List[str]								=			self.decode(
																								    encodings_and_masks=__lowerCamelCase   ,      input_tokens=__lowerCamelCase   ,      noise_time=t / self.scheduler.config.num_train_timesteps   ,      )

																								# Compute previous output: x_t -> x_t-1
																								A__       :     str								=			self.scheduler.step(__lowerCamelCase   ,      __lowerCamelCase   ,      __lowerCamelCase   ,      generator=__lowerCamelCase							).prev_sample

																	A__       :     str								=			self.scale_to_features(__lowerCamelCase   ,      input_range=[-1.0, 1.0]							)
																	A__       :     int								=			mel[:1]
																	A__       :     Union[str, Any]								=			mel.cpu().float().numpy()

																	A__       :     Any								=			np.concatenate([full_pred_mel, pred_mel[:1]]   ,      axis=1							)

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

																	logger.info("""Generated segment"""   ,      __lowerCamelCase							)

										if output_type == "numpy" and not is_onnx_available():
																	raise ValueError(
																	    """Cannot return output in \'np\' format if ONNX is not available. Make sure to have ONNX installed or set \'output_type\' to \'mel\'."""							)
										elif output_type == "numpy" and self.melgan is None:
																	raise ValueError(
																	    """Cannot return output in \'np\' format if melgan component is not defined. Make sure to define `self.melgan` or set \'output_type\' to \'mel\'."""							)

										if output_type == "numpy":
																	A__       :     Optional[int]								=			self.melgan(input_features=full_pred_mel.astype(np.floataa							)							)
										else:
																	A__       :     Optional[Any]								=			full_pred_mel

										if not return_dict:
																	return (output,)

										return AudioPipelineOutput(audios=__lowerCamelCase							)









'''simple docstring'''





import warnings
from typing import List

import numpy as np

from ...processing_utils import ProcessorMixin
from ...tokenization_utils_base import BatchEncoding
from ...utils import is_flax_available, is_tf_available, is_torch_available




class 				UpperCAmelCase			(			_snake_case						):
 UpperCAmelCase				=					["image_processor", "tokenizer"]
 UpperCAmelCase				=					"OwlViTImageProcessor"
 UpperCAmelCase				=					("CLIPTokenizer", "CLIPTokenizerFast")


 def __init__(      self							:	List[str]			,       __lowerCamelCase							:	Optional[int]=None			,       __lowerCamelCase							:	Optional[Any]=None			,       **__lowerCamelCase							:	str ):
    UpperCAmelCase__	:int      =      None
    if "feature_extractor" in kwargs:
       warnings.warn(
           '''The `feature_extractor` argument is deprecated and will be removed in v5, use `image_processor`'''
           ''' instead.'''			,       __lowerCamelCase			,       )
       UpperCAmelCase__	:Union[str, Any]      =      kwargs.pop('''feature_extractor''' )

    UpperCAmelCase__	:List[str]      =      image_processor if image_processor is not None else feature_extractor
    if image_processor is None:
       raise ValueError('''You need to specify an `image_processor`.''' )
    if tokenizer is None:
       raise ValueError('''You need to specify a `tokenizer`.''' )

    super().__init__(__lowerCamelCase			,       __lowerCamelCase )


 def __call__(      self							:	int			,       __lowerCamelCase							:	str=None			,       __lowerCamelCase							:	List[Any]=None			,       __lowerCamelCase							:	Dict=None			,       __lowerCamelCase							:	str="max_length"			,       __lowerCamelCase							:	Any="np"			,       **__lowerCamelCase							:	Tuple ):

    if text is None and query_images is None and images is None:
       raise ValueError(
           '''You have to specify at least one text or query image or image. All three cannot be none.''' )

    if text is not None:
       if isinstance(__lowerCamelCase			,       __lowerCamelCase ) or (isinstance(__lowerCamelCase			,       __lowerCamelCase ) and not isinstance(text[0]			,       __lowerCamelCase )):
          UpperCAmelCase__	:Any      =      [self.tokenizer(__lowerCamelCase			,       padding=__lowerCamelCase			,       return_tensors=__lowerCamelCase			,       **__lowerCamelCase )]

       elif isinstance(__lowerCamelCase			,       __lowerCamelCase ) and isinstance(text[0]			,       __lowerCamelCase ):
          UpperCAmelCase__	:Tuple      =      []

          # Maximum number of queries across batch
          UpperCAmelCase__	:List[str]      =      max([len(__lowerCamelCase ) for t in text] )

          # Pad all batch samples to max number of text queries
          for t in text:
             if len(__lowerCamelCase ) != max_num_queries:
                UpperCAmelCase__	:str      =      t + [''' '''] * (max_num_queries - len(__lowerCamelCase ))

             UpperCAmelCase__	:Tuple      =      self.tokenizer(__lowerCamelCase			,       padding=__lowerCamelCase			,       return_tensors=__lowerCamelCase			,       **__lowerCamelCase )
             encodings.append(__lowerCamelCase )
       else:
          raise TypeError('''Input text should be a string, a list of strings or a nested list of strings''' )

       if return_tensors == "np":
          UpperCAmelCase__	:List[Any]      =      np.concatenate([encoding['''input_ids'''] for encoding in encodings]			,       axis=0 )
          UpperCAmelCase__	:Any      =      np.concatenate([encoding['''attention_mask'''] for encoding in encodings]			,       axis=0 )

       elif return_tensors == "jax" and is_flax_available():
          import jax.numpy as jnp

          UpperCAmelCase__	:List[Any]      =      jnp.concatenate([encoding['''input_ids'''] for encoding in encodings]			,       axis=0 )
          UpperCAmelCase__	:Union[str, Any]      =      jnp.concatenate([encoding['''attention_mask'''] for encoding in encodings]			,       axis=0 )

       elif return_tensors == "pt" and is_torch_available():
          import torch

          UpperCAmelCase__	:Any      =      torch.cat([encoding['''input_ids'''] for encoding in encodings]			,       dim=0 )
          UpperCAmelCase__	:List[str]      =      torch.cat([encoding['''attention_mask'''] for encoding in encodings]			,       dim=0 )

       elif return_tensors == "tf" and is_tf_available():
          import tensorflow as tf

          UpperCAmelCase__	:Optional[Any]      =      tf.stack([encoding['''input_ids'''] for encoding in encodings]			,       axis=0 )
          UpperCAmelCase__	:int      =      tf.stack([encoding['''attention_mask'''] for encoding in encodings]			,       axis=0 )

       else:
          raise ValueError('''Target return tensor type could not be returned''' )

       UpperCAmelCase__	:List[Any]      =      BatchEncoding()
       UpperCAmelCase__	:Union[str, Any]      =      input_ids
       UpperCAmelCase__	:Dict      =      attention_mask

    if query_images is not None:
       UpperCAmelCase__	:Tuple      =      BatchEncoding()
       UpperCAmelCase__	:int      =      self.image_processor(
           __lowerCamelCase			,       return_tensors=__lowerCamelCase			,       **__lowerCamelCase ).pixel_values
       UpperCAmelCase__	:Optional[int]      =      query_pixel_values

    if images is not None:
       UpperCAmelCase__	:str      =      self.image_processor(__lowerCamelCase			,       return_tensors=__lowerCamelCase			,       **__lowerCamelCase )

    if text is not None and images is not None:
       UpperCAmelCase__	:Dict      =      image_features.pixel_values
       return encoding
    elif query_images is not None and images is not None:
       UpperCAmelCase__	:Tuple      =      image_features.pixel_values
       return encoding
    elif text is not None or query_images is not None:
       return encoding
    else:
       return BatchEncoding(data=dict(**__lowerCamelCase )			,       tensor_type=__lowerCamelCase )


 def        __SCREAMING_SNAKE_CASE					(      self							:	int			,       *__lowerCamelCase							:	Optional[int]			,       **__lowerCamelCase							:	int ):
    return self.image_processor.post_process(*__lowerCamelCase			,       **__lowerCamelCase )


 def        __SCREAMING_SNAKE_CASE					(      self							:	Dict			,       *__lowerCamelCase							:	Any			,       **__lowerCamelCase							:	Tuple ):
    return self.image_processor.post_process_object_detection(*__lowerCamelCase			,       **__lowerCamelCase )


 def        __SCREAMING_SNAKE_CASE					(      self							:	Union[str, Any]			,       *__lowerCamelCase							:	Tuple			,       **__lowerCamelCase							:	str ):
    return self.image_processor.post_process_image_guided_detection(*__lowerCamelCase			,       **__lowerCamelCase )


 def        __SCREAMING_SNAKE_CASE					(      self							:	int			,       *__lowerCamelCase							:	Optional[Any]			,       **__lowerCamelCase							:	Union[str, Any] ):
    return self.tokenizer.batch_decode(*__lowerCamelCase			,       **__lowerCamelCase )


 def        __SCREAMING_SNAKE_CASE					(      self							:	Optional[Any]			,       *__lowerCamelCase							:	Union[str, Any]			,       **__lowerCamelCase							:	Dict ):
    return self.tokenizer.decode(*__lowerCamelCase			,       **__lowerCamelCase )


 @property
 def        __SCREAMING_SNAKE_CASE					(      self							:	Union[str, Any] ):
    warnings.warn(
        '''`feature_extractor_class` is deprecated and will be removed in v5. Use `image_processor_class` instead.'''			,       __lowerCamelCase			,       )
    return self.image_processor_class





 @property
 def        __SCREAMING_SNAKE_CASE					(      self							:	Union[str, Any] ):
    warnings.warn(
        '''`feature_extractor` is deprecated and will be removed in v5. Use `image_processor` instead.'''			,       __lowerCamelCase			,       )
    return self.image_processor



