





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    :str							,				__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    :Any							,				__A    :Tuple							,				__A    :Tuple=(-1.0, 1.0)							,				__A    :Tuple=False		)   ->   Union[str, 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    :Any							,				__A    :Union[str, Any]							,				__A    :Any=(-1.0, 1.0)							,				__A    :Optional[int]=False		)   ->   List[str]:



									"""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    :Any							,				__A    :Tuple							,				__A    :List[Any]							,				__A    :str		)   ->   Tuple:



									"""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    :Union[str, Any]							,				__A    :List[Any]							,				__A    :Optional[Any]							,				__A    :Any		)   ->   Dict:



									"""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    :Tuple							,				__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		)






from ...configuration_utils import PretrainedConfig
from ...utils import logging
from ...utils.backbone_utils import BackboneConfigMixin, get_aligned_output_features_output_indices


_lowerCamelCase   				=			logging.get_logger(__name__)

_lowerCamelCase   				=			{
    'shi-labs/nat-mini-in1k-224': 'https://huggingface.co/shi-labs/nat-mini-in1k-224/resolve/main/config.json',
    # See all Nat models at https://huggingface.co/models?filter=nat
}




class   UpperCamelCase_       (     UpperCamelCase__				,  UpperCamelCase__    ):
							lowerCamelCase_           =					"nat"

							lowerCamelCase_           =					{
							    "num_attention_heads": "num_heads",
							    "num_hidden_layers": "num_layers",
							}


							def __init__(							self    :List[Any]							,				__A    :Optional[Any]=4							,				__A    :Any=3							,				__A    :Optional[int]=64							,				__A    :Optional[int]=[3, 4, 6, 5]							,				__A    :Union[str, Any]=[2, 4, 8, 16]							,				__A    :Optional[Any]=7							,				__A    :Optional[Any]=3.0							,				__A    :List[Any]=True							,				__A    :int=0.0							,				__A    :Dict=0.0							,				__A    :Optional[Any]=0.1							,				__A    :str="gelu"							,				__A    :Optional[Any]=0.0_2							,				__A    :Optional[int]=1E-5							,				__A    :Optional[int]=0.0							,				__A    :Optional[Any]=None							,				__A    :Union[str, Any]=None							,				**__A    :Union[str, Any]							,				)   ->   Optional[int]:



									"""simple docstring"""



									super().__init__(**__A		)

									SCREAMING_SNAKE_CASE__			=      patch_size
									SCREAMING_SNAKE_CASE__			=      num_channels
									SCREAMING_SNAKE_CASE__			=      embed_dim
									SCREAMING_SNAKE_CASE__			=      depths
									SCREAMING_SNAKE_CASE__			=      len(__A		)
									SCREAMING_SNAKE_CASE__			=      num_heads
									SCREAMING_SNAKE_CASE__			=      kernel_size
									SCREAMING_SNAKE_CASE__			=      mlp_ratio
									SCREAMING_SNAKE_CASE__			=      qkv_bias
									SCREAMING_SNAKE_CASE__			=      hidden_dropout_prob
									SCREAMING_SNAKE_CASE__			=      attention_probs_dropout_prob
									SCREAMING_SNAKE_CASE__			=      drop_path_rate
									SCREAMING_SNAKE_CASE__			=      hidden_act
									SCREAMING_SNAKE_CASE__			=      layer_norm_eps
									SCREAMING_SNAKE_CASE__			=      initializer_range
									# we set the hidden_size attribute in order to make Nat work with VisionEncoderDecoderModel
									# this indicates the channel dimension after the last stage of the model
									SCREAMING_SNAKE_CASE__			=      int(embed_dim * 2 ** (len(__A		) - 1)		)
									SCREAMING_SNAKE_CASE__			=      layer_scale_init_value
									SCREAMING_SNAKE_CASE__			=      ["""stem"""] + [f'''stage{idx}''' for idx in range(1							,				len(__A		) + 1		)]
									SCREAMING_SNAKE_CASE__	,						SCREAMING_SNAKE_CASE__			=      get_aligned_output_features_output_indices(
									    out_features=__A							,				out_indices=__A							,				stage_names=self.stage_names		)