




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


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

__A					=      256
class       lowerCamelCase__     (	_UpperCamelCase      ):
			a__			:   Union[str, Any]             =  ["""melgan"""]


			def __init__(  self					,			SCREAMING_SNAKE_CASE					,			SCREAMING_SNAKE_CASE					,			SCREAMING_SNAKE_CASE					,			SCREAMING_SNAKE_CASE					,			SCREAMING_SNAKE_CASE					,			):



					"""simple docstring"""







					super().__init__()

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

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


			def 				lowerCamelCase_					(  self					,			SCREAMING_SNAKE_CASE					,			SCREAMING_SNAKE_CASE=(-1.0, 1.0)					,			SCREAMING_SNAKE_CASE=False				):



					"""simple docstring"""







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



					"""simple docstring"""







					snake_case		:    Any             =					input_range
					snake_case		:    List[Any]             =					torch.clip(lowerCamelCase_					,			lowerCamelCase_					,			lowerCamelCase_				) if clip else outputs
					# Scale to [0, 1].
					snake_case		:    Optional[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					,			SCREAMING_SNAKE_CASE					,			SCREAMING_SNAKE_CASE					,			SCREAMING_SNAKE_CASE				):



					"""simple docstring"""







					snake_case		:    List[Any]             =					input_tokens > 0
					snake_case		:    Optional[int]             =					self.notes_encoder(
					    encoder_input_tokens=lowerCamelCase_					,			encoder_inputs_mask=lowerCamelCase_				)

					snake_case		:    Optional[int]             =					self.continuous_encoder(
					    encoder_inputs=lowerCamelCase_					,			encoder_inputs_mask=lowerCamelCase_				)

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


			def 				lowerCamelCase_					(  self					,			SCREAMING_SNAKE_CASE					,			SCREAMING_SNAKE_CASE					,			SCREAMING_SNAKE_CASE				):



					"""simple docstring"""







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

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

					snake_case		:    List[str]             =					self.decoder(
					    encodings_and_masks=lowerCamelCase_					,			decoder_input_tokens=lowerCamelCase_					,			decoder_noise_time=lowerCamelCase_				)
					return logits



			@torch.no_grad()
			def __call__(  self					,			SCREAMING_SNAKE_CASE					,			SCREAMING_SNAKE_CASE = None					,			SCREAMING_SNAKE_CASE = 100					,			SCREAMING_SNAKE_CASE = True					,			SCREAMING_SNAKE_CASE = "numpy"					,			SCREAMING_SNAKE_CASE = None					,			SCREAMING_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_				)}.'''				)

					snake_case		:    str             =					np.zeros([1, TARGET_FEATURE_LENGTH, self.n_dims]					,			dtype=np.floataa				)
					snake_case		:    Dict             =					np.zeros([1, 0, self.n_dims]					,			np.floataa				)
					snake_case		:    str             =					torch.ones((1, TARGET_FEATURE_LENGTH)					,			dtype=lowerCamelCase_					,			device=self.device				)

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

							snake_case		:    Any             =					self.scale_features(
							    lowerCamelCase_					,			output_range=[-1.0, 1.0]					,			clip=lowerCamelCase_				)

							snake_case		:    List[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
							snake_case		:    List[str]             =					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				)				):
									snake_case		:    Optional[Any]             =					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
									snake_case		:    int             =					self.scheduler.step(lowerCamelCase_					,			lowerCamelCase_					,			lowerCamelCase_					,			generator=lowerCamelCase_				).prev_sample

							snake_case		:    str             =					self.scale_to_features(lowerCamelCase_					,			input_range=[-1.0, 1.0]				)
							snake_case		:    int             =					mel[:1]
							snake_case		:    Dict             =					mel.cpu().float().numpy()

							snake_case		:    Optional[int]             =					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":
							snake_case		:    Optional[int]             =					self.melgan(input_features=full_pred_mel.astype(np.floataa				)				)
					else:
							snake_case		:    str             =					full_pred_mel

					if not return_dict:
							return (output,)

					return AudioPipelineOutput(audios=lowerCamelCase_				)








from math import ceil
from typing import List, Optional, Union

import numpy as np

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


__UpperCamelCase:	Optional[int] = logging.get_logger(__name__)


class       __lowerCAmelCase					(       _UpperCamelCase      ):


	'''simple docstring'''





	_A													=				["audio_values", "audio_mask"]

	def __init__(						self:						Any,					lowerCamelCase_:						int=2048,					lowerCamelCase_:						str=1,					lowerCamelCase_:						str=[16, 16],					lowerCamelCase_:						Union[str, Any]=128,					lowerCamelCase_:						Tuple=44100,					lowerCamelCase_:						Union[str, Any]=86,					lowerCamelCase_:						Optional[Any]=2048,					lowerCamelCase_:						Dict=0.0,					**lowerCamelCase_:						str,					):
			super().__init__(
			    feature_size=lowerCamelCase_,					sampling_rate=lowerCamelCase_,					padding_value=lowerCamelCase_,					**lowerCamelCase_,					)

			lowercase__					:      List[str]      =		spectrogram_length
			lowercase__					:      Union[str, Any]      =		num_channels
			lowercase__					:      int      =		patch_size
			lowercase__					:      Union[str, Any]      =		feature_size // self.patch_size[1]
			lowercase__					:      Optional[int]      =		n_fft
			lowercase__					:      int      =		sampling_rate // hop_length_to_sampling_rate
			lowercase__					:      str      =		sampling_rate
			lowercase__					:      Union[str, Any]      =		padding_value
			lowercase__					:      Any      =		mel_filter_bank(
			    num_frequency_bins=1 + n_fft // 2,					num_mel_filters=lowerCamelCase_,					min_frequency=0.0,					max_frequency=2_2_0_5_0.0,					sampling_rate=lowerCamelCase_,					norm='slaney',					mel_scale='slaney',					).T

	def    snake_case__(						self:						Tuple,					lowerCamelCase_:						np.array			):
			lowercase__					:      Optional[Any]      =		spectrogram(
			    lowerCamelCase_,					window_function(self.n_fft,					'hann'			),					frame_length=self.n_fft,					hop_length=self.hop_length,					power=2.0,					mel_filters=self.mel_filters.T,					log_mel='dB',					db_range=8_0.0,					)
			lowercase__					:      Optional[Any]      =		log_spec[:, :-1]
			lowercase__					:      int      =		log_spec - 2_0.0
			lowercase__					:      Dict      =		np.clip(log_spec / 4_0.0,					-2.0,					0.0			) + 1.0
			return log_spec





	def __call__(						self:						Dict,					lowerCamelCase_:						Union[np.ndarray, List[float], List[np.ndarray], List[List[float]]],					lowerCamelCase_:						Optional[Union[str, TensorType]] = None,					lowerCamelCase_:						Optional[bool] = True,					lowerCamelCase_:						Optional[int] = None,					lowerCamelCase_:						bool = False,					lowerCamelCase_:						bool = False,					**lowerCamelCase_:						str,					):

			if sampling_rate is not None:
					if sampling_rate != self.sampling_rate:
							raise ValueError(
							    'This feature extractor is set to support sampling rate'
							    F""" of {self.sampling_rate}. Please make sure that the provided `raw_speech` input was sampled"""
							    F""" 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.'			)

			lowercase__					:      Any      =		isinstance(lowerCamelCase_,					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}"""			)
			lowercase__					:      Optional[Any]      =		is_batched_numpy or (
			    isinstance(lowerCamelCase_,					(list, tuple)			) and (isinstance(raw_speech[0],					(np.ndarray, tuple, list)			))
			)
			if is_batched:
					lowercase__					:      int      =		[np.asarray([speech],					dtype=np.floataa			).T for speech in raw_speech]
			elif not is_batched and not isinstance(lowerCamelCase_,					np.ndarray			):
					lowercase__					:      Optional[int]      =		np.asarray(lowerCamelCase_,					dtype=np.floataa			)
			elif isinstance(lowerCamelCase_,					np.ndarray			) and raw_speech.dtype is np.dtype(np.floataa			):
					lowercase__					:      str      =		raw_speech.astype(np.floataa			)
			# always return batch
			if not is_batched:
					lowercase__					:      Tuple      =		[np.asarray([raw_speech]			).T]

			# Convert audio signals to log mel spectrograms, truncate by time axis
			lowercase__					:      List[str]      =		[
			    self._np_extract_fbank_features(waveform.squeeze()			).T[: self.spectrogram_length] for waveform in raw_speech
			]
			if isinstance(audio_features[0],					lowerCamelCase_			):
					lowercase__					:      str      =		[np.asarray(lowerCamelCase_,					dtype=np.floataa			) for feature in audio_features]

			# Create audio attention mask
			lowercase__					:      Dict      =		max(
			    [ceil(feature.shape[0] / self.patch_size[0]			) * self.freq_len for feature in audio_features]			)  # The maximum number of audio patches in a batch
			if return_attention_mask:
					lowercase__					:      int      =		[
					    (ceil(feature.shape[0] / self.patch_size[0]			) * self.freq_len) * [1]
					    + (max_patch_len - ceil(feature.shape[0] / self.patch_size[0]			) * self.freq_len) * [0]
					    for feature in audio_features
					]
					lowercase__					:      Any      =		np.array(lowerCamelCase_			).astype(np.floataa			)

			# convert into correct format for padding
			lowercase__					:      Any      =		max_patch_len // self.freq_len * self.patch_size[0]  # The maximum audio size in a batch
			lowercase__					:      Tuple      =		np.ones([len(lowerCamelCase_			), 1, max_time_len, self.feature_size]			).astype(np.floataa			)
			lowercase__					:      Dict      =		padded_audio_features * self.padding_value
			for i in range(len(lowerCamelCase_			)			):
					lowercase__					:      Union[str, Any]      =		audio_features[i]
					lowercase__					:      str      =		feature

			# return as BatchFeature
			if return_attention_mask:
					lowercase__					:      Dict      =		{'audio_values': padded_audio_features, 'audio_mask': audio_mask}
			else:
					lowercase__					:      int      =		{'audio_values': padded_audio_features}

			lowercase__					:      List[Any]      =		BatchFeature(data=lowerCamelCase_,					tensor_type=lowerCamelCase_			)
			return encoded_inputs


