





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


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

_lowerCamelCase 					=      256




class 	UpperCamelCase_							(  UpperCamelCase__     ):
							lowerCamelCase_							     =       ["melgan"]


							def __init__(				self     :List[Any]				,					__A     :SpectrogramNotesEncoder				,					__A     :SpectrogramContEncoder				,					__A     :TaFilmDecoder				,					__A     :DDPMScheduler				,					__A     :OnnxRuntimeModel if is_onnx_available() else Any				,					) -> None:



									"""simple docstring"""



									super().__init__()

									# From MELGAN
									SCREAMING_SNAKE_CASE__								=   math.log(1E-5		)  # Matches MelGAN training.
									SCREAMING_SNAKE_CASE__								=   4.0  # Largest value for most examples
									SCREAMING_SNAKE_CASE__								=   128

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


							def   _snake_case	(				self     :str				,					__A     :List[Any]				,					__A     :Optional[int]=(-1.0, 1.0)				,					__A     :Optional[Any]=False		) -> Optional[Any]:



									"""simple docstring"""



									SCREAMING_SNAKE_CASE__						,    SCREAMING_SNAKE_CASE__								=   output_range
									if clip:
											SCREAMING_SNAKE_CASE__								=   torch.clip(__A				,					self.min_value				,					self.max_value		)
									# Scale to [0, 1].
									SCREAMING_SNAKE_CASE__								=   (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   _snake_case	(				self     :Dict				,					__A     :Tuple				,					__A     :str=(-1.0, 1.0)				,					__A     :List[str]=False		) -> Tuple:



									"""simple docstring"""



									SCREAMING_SNAKE_CASE__						,    SCREAMING_SNAKE_CASE__								=   input_range
									SCREAMING_SNAKE_CASE__								=   torch.clip(__A				,					__A				,					__A		) if clip else outputs
									# Scale to [0, 1].
									SCREAMING_SNAKE_CASE__								=   (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   _snake_case	(				self     :Union[str, Any]				,					__A     :Any				,					__A     :List[Any]				,					__A     :str		) -> List[str]:



									"""simple docstring"""



									SCREAMING_SNAKE_CASE__								=   input_tokens > 0
									SCREAMING_SNAKE_CASE__						,    SCREAMING_SNAKE_CASE__								=   self.notes_encoder(
									    encoder_input_tokens=__A				,					encoder_inputs_mask=__A		)

									SCREAMING_SNAKE_CASE__						,    SCREAMING_SNAKE_CASE__								=   self.continuous_encoder(
									    encoder_inputs=__A				,					encoder_inputs_mask=__A		)

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


							def   _snake_case	(				self     :Any				,					__A     :int				,					__A     :str				,					__A     :Dict		) -> List[str]:



									"""simple docstring"""



									SCREAMING_SNAKE_CASE__								=   noise_time
									if not torch.is_tensor(__A		):
											SCREAMING_SNAKE_CASE__								=   torch.tensor([timesteps]				,					dtype=torch.long				,					device=input_tokens.device		)
									elif torch.is_tensor(__A		) and len(timesteps.shape		) == 0:
											SCREAMING_SNAKE_CASE__								=   timesteps[None].to(input_tokens.device		)

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

									SCREAMING_SNAKE_CASE__								=   self.decoder(
									    encodings_and_masks=__A				,					decoder_input_tokens=__A				,					decoder_noise_time=__A		)
									return logits


							@torch.no_grad()
							def __call__(				self     :Dict				,					__A     :List[List[int]]				,					__A     :Optional[torch.Generator] = None				,					__A     :int = 100				,					__A     :bool = True				,					__A     :str = "numpy"				,					__A     :Optional[Callable[[int, int, torch.FloatTensor], None]] = None				,					__A     :int = 1				,					) -> Union[AudioPipelineOutput, Tuple]:



									"""simple docstring"""



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

									SCREAMING_SNAKE_CASE__								=   np.zeros([1, TARGET_FEATURE_LENGTH, self.n_dims]				,					dtype=np.floataa		)
									SCREAMING_SNAKE_CASE__								=   np.zeros([1, 0, self.n_dims]				,					np.floataa		)
									SCREAMING_SNAKE_CASE__								=   torch.ones((1, TARGET_FEATURE_LENGTH)				,					dtype=__A				,					device=self.device		)

									for i, encoder_input_tokens in enumerate(__A		):
											if i == 0:
													SCREAMING_SNAKE_CASE__								=   torch.from_numpy(pred_mel[:1].copy()		).to(
													    device=self.device				,					dtype=self.decoder.dtype		)
													# The first chunk has no previous context.
													SCREAMING_SNAKE_CASE__								=   torch.zeros((1, TARGET_FEATURE_LENGTH)				,					dtype=__A				,					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.
													SCREAMING_SNAKE_CASE__								=   ones

											SCREAMING_SNAKE_CASE__								=   self.scale_features(
											    __A				,					output_range=[-1.0, 1.0]				,					clip=__A		)

											SCREAMING_SNAKE_CASE__								=   self.encode(
											    input_tokens=torch.IntTensor([encoder_input_tokens]		).to(device=self.device		)				,					continuous_inputs=__A				,					continuous_mask=__A				,					)

											# Sample encoder_continuous_inputs shaped gaussian noise to begin loop
											SCREAMING_SNAKE_CASE__								=   randn_tensor(
											    shape=encoder_continuous_inputs.shape				,					generator=__A				,					device=self.device				,					dtype=self.decoder.dtype				,					)

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

											# Denoising diffusion loop
											for j, t in enumerate(self.progress_bar(self.scheduler.timesteps		)		):
													SCREAMING_SNAKE_CASE__								=   self.decode(
													    encodings_and_masks=__A				,					input_tokens=__A				,					noise_time=t / self.scheduler.config.num_train_timesteps				,					)

													# Compute previous output: x_t -> x_t-1
													SCREAMING_SNAKE_CASE__								=   self.scheduler.step(__A				,					__A				,					__A				,					generator=__A		).prev_sample

											SCREAMING_SNAKE_CASE__								=   self.scale_to_features(__A				,					input_range=[-1.0, 1.0]		)
											SCREAMING_SNAKE_CASE__								=   mel[:1]
											SCREAMING_SNAKE_CASE__								=   mel.cpu().float().numpy()

											SCREAMING_SNAKE_CASE__								=   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(__A				,					__A		)

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

									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":
											SCREAMING_SNAKE_CASE__								=   self.melgan(input_features=full_pred_mel.astype(np.floataa		)		)
									else:
											SCREAMING_SNAKE_CASE__								=   full_pred_mel

									if not return_dict:
											return (output,)

									return AudioPipelineOutput(audios=__A		)






import copy

from ...configuration_utils import PretrainedConfig
from ...utils import logging


_lowerCamelCase 					=      logging.get_logger(__name__)




class 	UpperCamelCase_							(  UpperCamelCase__     ):
							lowerCamelCase_							     =       "encoder-decoder"
							lowerCamelCase_							     =       True


							def __init__(				self     :Optional[int]				,					**__A     :str		) -> int:



									"""simple docstring"""



									super().__init__(**__A		)
									assert (
									    "encoder" in kwargs and "decoder" in kwargs
									), "Config has to be initialized with encoder and decoder config"
									SCREAMING_SNAKE_CASE__								=   kwargs.pop("""encoder"""		)
									SCREAMING_SNAKE_CASE__								=   encoder_config.pop("""model_type"""		)
									SCREAMING_SNAKE_CASE__								=   kwargs.pop("""decoder"""		)
									SCREAMING_SNAKE_CASE__								=   decoder_config.pop("""model_type"""		)

									from ..auto.configuration_auto import AutoConfig

									SCREAMING_SNAKE_CASE__								=   AutoConfig.for_model(__A				,					**__A		)
									SCREAMING_SNAKE_CASE__								=   AutoConfig.for_model(__A				,					**__A		)
									SCREAMING_SNAKE_CASE__								=   True


							@classmethod
							def   _snake_case	(				cls     :str				,					__A     :PretrainedConfig				,					__A     :PretrainedConfig				,					**__A     :List[str]		) -> PretrainedConfig:



									"""simple docstring"""



									logger.info("""Set `config.is_decoder=True` and `config.add_cross_attention=True` for decoder_config"""		)
									SCREAMING_SNAKE_CASE__								=   True
									SCREAMING_SNAKE_CASE__								=   True

									return cls(encoder=encoder_config.to_dict()				,					decoder=decoder_config.to_dict()				,					**__A		)


							def   _snake_case	(				self     :str		) -> Union[str, Any]:



									"""simple docstring"""



									SCREAMING_SNAKE_CASE__								=   copy.deepcopy(self.__dict__		)
									SCREAMING_SNAKE_CASE__								=   self.encoder.to_dict()
									SCREAMING_SNAKE_CASE__								=   self.decoder.to_dict()
									SCREAMING_SNAKE_CASE__								=   self.__class__.model_type
									return output