



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


lowerCAmelCase_     :       Optional[Any]								=						logging.get_logger(__name__)  # pylint: disable=invalid-name

lowerCAmelCase_     :       Union[str, Any]								=						256
class 	lowerCamelCase_							(					snake_case_			):
      _lowerCAmelCase     :   Tuple	     =				['melgan']







      def __init__(  self   :      Dict				,    lowerCAmelCase__   :      SpectrogramNotesEncoder				,    lowerCAmelCase__   :      SpectrogramContEncoder				,    lowerCAmelCase__   :      TaFilmDecoder				,    lowerCAmelCase__   :      DDPMScheduler				,    lowerCAmelCase__   :      OnnxRuntimeModel if is_onnx_available() else Any				,    ):

             """simple docstring"""
             super().__init__()

             # From MELGAN
             SCREAMING_SNAKE_CASE					:      List[Any]     					=			math.log(1e-5				)  # Matches MelGAN training.
             SCREAMING_SNAKE_CASE					:      Tuple     					=			4.0  # Largest value for most examples
             SCREAMING_SNAKE_CASE					:      int     					=			1_28

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







      def 					__lowercase    (  self   :      List[str]				,    lowerCAmelCase__   :      str				,    lowerCAmelCase__   :      Any=(-1.0, 1.0)				,    lowerCAmelCase__   :      Optional[int]=False				):

             """simple docstring"""
             SCREAMING_SNAKE_CASE					:      List[Any]     					=			output_range
             if clip:
                    SCREAMING_SNAKE_CASE					:      str     					=			torch.clip(lowerCAmelCase__				,    self.min_value				,    self.max_value				)
             # Scale to [0, 1].
             SCREAMING_SNAKE_CASE					:      Union[str, Any]     					=			(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 					__lowercase    (  self   :      str				,    lowerCAmelCase__   :      Optional[Any]				,    lowerCAmelCase__   :      int=(-1.0, 1.0)				,    lowerCAmelCase__   :      Union[str, Any]=False				):

             """simple docstring"""
             SCREAMING_SNAKE_CASE					:      int     					=			input_range
             SCREAMING_SNAKE_CASE					:      Union[str, Any]     					=			torch.clip(lowerCAmelCase__				,    lowerCAmelCase__				,    lowerCAmelCase__				) if clip else outputs
             # Scale to [0, 1].
             SCREAMING_SNAKE_CASE					:      Tuple     					=			(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 					__lowercase    (  self   :      List[Any]				,    lowerCAmelCase__   :      Any				,    lowerCAmelCase__   :      Tuple				,    lowerCAmelCase__   :      Optional[Any]				):

             """simple docstring"""
             SCREAMING_SNAKE_CASE					:      Optional[Any]     					=			input_tokens > 0
             SCREAMING_SNAKE_CASE					:      List[str]     					=			self.notes_encoder(
                 encoder_input_tokens=lowerCAmelCase__				,    encoder_inputs_mask=lowerCAmelCase__				)

             SCREAMING_SNAKE_CASE					:      Dict     					=			self.continuous_encoder(
                 encoder_inputs=lowerCAmelCase__				,    encoder_inputs_mask=lowerCAmelCase__				)

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







      def 					__lowercase    (  self   :      int				,    lowerCAmelCase__   :      str				,    lowerCAmelCase__   :      List[str]				,    lowerCAmelCase__   :      List[str]				):

             """simple docstring"""
             SCREAMING_SNAKE_CASE					:      List[Any]     					=			noise_time
             if not torch.is_tensor(lowerCAmelCase__				):
                    SCREAMING_SNAKE_CASE					:      Union[str, Any]     					=			torch.tensor([timesteps]				,    dtype=torch.long				,    device=input_tokens.device				)
             elif torch.is_tensor(lowerCAmelCase__				) and len(timesteps.shape				) == 0:
                    SCREAMING_SNAKE_CASE					:      Optional[Any]     					=			timesteps[None].to(input_tokens.device				)

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

             SCREAMING_SNAKE_CASE					:      int     					=			self.decoder(
                 encodings_and_masks=lowerCAmelCase__				,    decoder_input_tokens=lowerCAmelCase__				,    decoder_noise_time=lowerCAmelCase__				)
             return logits







      @torch.no_grad()
      def __call__(  self   :      Optional[int]				,    lowerCAmelCase__   :      List[List[int]]				,    lowerCAmelCase__   :      Optional[torch.Generator] = None				,    lowerCAmelCase__   :      int = 1_00				,    lowerCAmelCase__   :      bool = True				,    lowerCAmelCase__   :      str = "numpy"				,    lowerCAmelCase__   :      Optional[Callable[[int, int, torch.FloatTensor], None]] = None				,    lowerCAmelCase__   :      int = 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__				)}."""				)

             SCREAMING_SNAKE_CASE					:      int     					=			np.zeros([1, TARGET_FEATURE_LENGTH, self.n_dims]				,    dtype=np.floataa				)
             SCREAMING_SNAKE_CASE					:      List[Any]     					=			np.zeros([1, 0, self.n_dims]				,    np.floataa				)
             SCREAMING_SNAKE_CASE					:      Any     					=			torch.ones((1, TARGET_FEATURE_LENGTH)				,    dtype=lowerCAmelCase__				,    device=self.device				)

             for i, encoder_input_tokens in enumerate(lowerCAmelCase__				):
                    if i == 0:
                           SCREAMING_SNAKE_CASE					:      str     					=			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					:      List[str]     					=			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.
                           SCREAMING_SNAKE_CASE					:      Dict     					=			ones

                    SCREAMING_SNAKE_CASE					:      str     					=			self.scale_features(
                        lowerCAmelCase__				,    output_range=[-1.0, 1.0]				,    clip=lowerCAmelCase__				)

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

                    SCREAMING_SNAKE_CASE					:      Optional[Any]     					=			self.scale_to_features(lowerCAmelCase__				,    input_range=[-1.0, 1.0]				)
                    SCREAMING_SNAKE_CASE					:      List[str]     					=			mel[:1]
                    SCREAMING_SNAKE_CASE					:      Any     					=			mel.cpu().float().numpy()

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

             if not return_dict:
                    return (output,)

             return AudioPipelineOutput(audios=lowerCAmelCase__				)









'''simple docstring'''



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


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

lowerCAmelCase_     :       Optional[Any]								=						{
    'microsoft/swinv2-tiny-patch4-window8-256': (
        'https://huggingface.co/microsoft/swinv2-tiny-patch4-window8-256/resolve/main/config.json'
    ),
}
class 	lowerCamelCase_							(					snake_case_			):
						_lowerCAmelCase     :   str	     =				'swinv2'

						_lowerCAmelCase     :   Optional[int]	     =				{
						    'num_attention_heads': 'num_heads',
						    'num_hidden_layers': 'num_layers',
						}







						def __init__(  self   :      Optional[int]				,    lowerCAmelCase__   :      Any=2_24				,    lowerCAmelCase__   :      Optional[Any]=4				,    lowerCAmelCase__   :      Tuple=3				,    lowerCAmelCase__   :      Tuple=96				,    lowerCAmelCase__   :      Optional[int]=[2, 2, 6, 2]				,    lowerCAmelCase__   :      Tuple=[3, 6, 12, 24]				,    lowerCAmelCase__   :      List[str]=7				,    lowerCAmelCase__   :      Tuple=4.0				,    lowerCAmelCase__   :      Dict=True				,    lowerCAmelCase__   :      Any=0.0				,    lowerCAmelCase__   :      Optional[int]=0.0				,    lowerCAmelCase__   :      Optional[Any]=0.1				,    lowerCAmelCase__   :      int="gelu"				,    lowerCAmelCase__   :      Tuple=False				,    lowerCAmelCase__   :      Tuple=0.02				,    lowerCAmelCase__   :      Dict=1e-5				,    lowerCAmelCase__   :      List[str]=32				,    **lowerCAmelCase__   :      Union[str, Any]				,    ):

													"""simple docstring"""
													super().__init__(**lowerCAmelCase__				)

													SCREAMING_SNAKE_CASE					:      int     					=			image_size
													SCREAMING_SNAKE_CASE					:      Optional[int]     					=			patch_size
													SCREAMING_SNAKE_CASE					:      Union[str, Any]     					=			num_channels
													SCREAMING_SNAKE_CASE					:      int     					=			embed_dim
													SCREAMING_SNAKE_CASE					:      List[Any]     					=			depths
													SCREAMING_SNAKE_CASE					:      str     					=			len(lowerCAmelCase__				)
													SCREAMING_SNAKE_CASE					:      Optional[Any]     					=			num_heads
													SCREAMING_SNAKE_CASE					:      Tuple     					=			window_size
													SCREAMING_SNAKE_CASE					:      Any     					=			mlp_ratio
													SCREAMING_SNAKE_CASE					:      Optional[Any]     					=			qkv_bias
													SCREAMING_SNAKE_CASE					:      Optional[Any]     					=			hidden_dropout_prob
													SCREAMING_SNAKE_CASE					:      int     					=			attention_probs_dropout_prob
													SCREAMING_SNAKE_CASE					:      List[str]     					=			drop_path_rate
													SCREAMING_SNAKE_CASE					:      Any     					=			hidden_act
													SCREAMING_SNAKE_CASE					:      List[Any]     					=			use_absolute_embeddings
													SCREAMING_SNAKE_CASE					:      List[str]     					=			layer_norm_eps
													SCREAMING_SNAKE_CASE					:      Union[str, Any]     					=			initializer_range
													SCREAMING_SNAKE_CASE					:      Dict     					=			encoder_stride
													# we set the hidden_size attribute in order to make Swinv2 work with VisionEncoderDecoderModel
													# this indicates the channel dimension after the last stage of the model
													SCREAMING_SNAKE_CASE					:      Optional[Any]     					=			int(embed_dim * 2 ** (len(lowerCAmelCase__				) - 1)				)
													SCREAMING_SNAKE_CASE					:      int     					=			(0, 0, 0, 0)




