


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_      : Optional[Any]  					=       logging.get_logger(__name__)  # pylint: disable=invalid-name

UpperCAmelCase_      : Union[str, Any]  					=       256
class  SCREAMING_SNAKE_CASE__  (				lowercase__			):
      snake_case__				:     Union[str, Any]        =						['''melgan''']
      def __init__(       self				:       Any     ,							SCREAMING_SNAKE_CASE__				:       SpectrogramNotesEncoder     ,							SCREAMING_SNAKE_CASE__				:       SpectrogramContEncoder     ,							SCREAMING_SNAKE_CASE__				:       TaFilmDecoder     ,							SCREAMING_SNAKE_CASE__				:       DDPMScheduler     ,							SCREAMING_SNAKE_CASE__				:       OnnxRuntimeModel if is_onnx_available() else Any     ,							)		->			None:
             super().__init__()

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

             self.register_modules(
                 notes_encoder=SCREAMING_SNAKE_CASE__     ,							continuous_encoder=SCREAMING_SNAKE_CASE__     ,							decoder=SCREAMING_SNAKE_CASE__     ,							scheduler=SCREAMING_SNAKE_CASE__     ,							melgan=SCREAMING_SNAKE_CASE__     ,							)
      def        SCREAMING_SNAKE_CASE							(       self				:       int     ,							SCREAMING_SNAKE_CASE__				:       List[str]     ,							SCREAMING_SNAKE_CASE__				:       Dict=(-1.0, 1.0)     ,							SCREAMING_SNAKE_CASE__				:       Union[str, Any]=False     )		->			List[str]:
             a_							,		a_					:		str							=  output_range
             if clip:
                    a_					:		Dict							=  torch.clip(SCREAMING_SNAKE_CASE__     ,							self.min_value     ,							self.max_value     )
             # Scale to [0, 1].
             a_					:		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        SCREAMING_SNAKE_CASE							(       self				:       Optional[Any]     ,							SCREAMING_SNAKE_CASE__				:       Dict     ,							SCREAMING_SNAKE_CASE__				:       str=(-1.0, 1.0)     ,							SCREAMING_SNAKE_CASE__				:       Union[str, Any]=False     )		->			str:
             a_							,		a_					:		str							=  input_range
             a_					:		Tuple							=  torch.clip(SCREAMING_SNAKE_CASE__     ,							SCREAMING_SNAKE_CASE__     ,							SCREAMING_SNAKE_CASE__     ) if clip else outputs
             # Scale to [0, 1].
             a_					:		str							=  (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        SCREAMING_SNAKE_CASE							(       self				:       Union[str, Any]     ,							SCREAMING_SNAKE_CASE__				:       List[str]     ,							SCREAMING_SNAKE_CASE__				:       str     ,							SCREAMING_SNAKE_CASE__				:       Any     )		->			Union[str, Any]:
             a_					:		Optional[Any]							=  input_tokens > 0
             a_							,		a_					:		Tuple							=  self.notes_encoder(
                 encoder_input_tokens=SCREAMING_SNAKE_CASE__     ,							encoder_inputs_mask=SCREAMING_SNAKE_CASE__     )

             a_							,		a_					:		Union[str, Any]							=  self.continuous_encoder(
                 encoder_inputs=SCREAMING_SNAKE_CASE__     ,							encoder_inputs_mask=SCREAMING_SNAKE_CASE__     )

             return [(tokens_encoded, tokens_mask), (continuous_encoded, continuous_mask)]
      def        SCREAMING_SNAKE_CASE							(       self				:       str     ,							SCREAMING_SNAKE_CASE__				:       Dict     ,							SCREAMING_SNAKE_CASE__				:       Optional[Any]     ,							SCREAMING_SNAKE_CASE__				:       List[str]     )		->			int:
             a_					:		Union[str, Any]							=  noise_time
             if not torch.is_tensor(SCREAMING_SNAKE_CASE__     ):
                    a_					:		int							=  torch.tensor([timesteps]     ,							dtype=torch.long     ,							device=input_tokens.device     )
             elif torch.is_tensor(SCREAMING_SNAKE_CASE__     ) and len(timesteps.shape     ) == 0:
                    a_					:		Dict							=  timesteps[None].to(input_tokens.device     )

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

             a_					:		Optional[Any]							=  self.decoder(
                 encodings_and_masks=SCREAMING_SNAKE_CASE__     ,							decoder_input_tokens=SCREAMING_SNAKE_CASE__     ,							decoder_noise_time=SCREAMING_SNAKE_CASE__     )
             return logits

      @torch.no_grad()
      def __call__(       self				:       Union[str, Any]     ,							SCREAMING_SNAKE_CASE__				:       List[List[int]]     ,							SCREAMING_SNAKE_CASE__				:       Optional[torch.Generator] = None     ,							SCREAMING_SNAKE_CASE__				:       int = 1_0_0     ,							SCREAMING_SNAKE_CASE__				:       bool = True     ,							SCREAMING_SNAKE_CASE__				:       str = "numpy"     ,							SCREAMING_SNAKE_CASE__				:       Optional[Callable[[int, int, torch.FloatTensor], None]] = None     ,							SCREAMING_SNAKE_CASE__				:       int = 1     ,							)		->			Union[AudioPipelineOutput, Tuple]:
             if (callback_steps is None) or (
                 callback_steps is not None and (not isinstance(SCREAMING_SNAKE_CASE__     ,							SCREAMING_SNAKE_CASE__     ) or callback_steps <= 0)
             ):
                    raise ValueError(
                        F"""`callback_steps` has to be a positive integer but is {callback_steps} of type"""
                        F""" {type(SCREAMING_SNAKE_CASE__     )}."""     )

             a_					:		Tuple							=  np.zeros([1, TARGET_FEATURE_LENGTH, self.n_dims]     ,							dtype=np.floataa     )
             a_					:		Any							=  np.zeros([1, 0, self.n_dims]     ,							np.floataa     )
             a_					:		str							=  torch.ones((1, TARGET_FEATURE_LENGTH)     ,							dtype=SCREAMING_SNAKE_CASE__     ,							device=self.device     )

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

                    a_					:		Union[str, Any]							=  self.scale_features(
                        SCREAMING_SNAKE_CASE__     ,							output_range=[-1.0, 1.0]     ,							clip=SCREAMING_SNAKE_CASE__     )

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

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

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

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

                           # Compute previous output: x_t -> x_t-1
                           a_					:		List[str]							=  self.scheduler.step(SCREAMING_SNAKE_CASE__     ,							SCREAMING_SNAKE_CASE__     ,							SCREAMING_SNAKE_CASE__     ,							generator=SCREAMING_SNAKE_CASE__     ).prev_sample

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

                    a_					:		List[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(SCREAMING_SNAKE_CASE__     ,							SCREAMING_SNAKE_CASE__     )

                    logger.info('Generated segment'     ,							SCREAMING_SNAKE_CASE__     )

             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_					:		List[Any]							=  self.melgan(input_features=full_pred_mel.astype(np.floataa     )     )
             else:
                    a_					:		Any							=  full_pred_mel

             if not return_dict:
                    return (output,)

             return AudioPipelineOutput(audios=SCREAMING_SNAKE_CASE__     )






from collections import OrderedDict
from typing import Mapping

from ...configuration_utils import PretrainedConfig
from ...onnx import OnnxConfig
from ...utils import logging


UpperCAmelCase_      : str  					=       logging.get_logger(__name__)

UpperCAmelCase_      : List[str]  					=       {
    'YituTech/conv-bert-base': 'https://huggingface.co/YituTech/conv-bert-base/resolve/main/config.json',
    'YituTech/conv-bert-medium-small': (
        'https://huggingface.co/YituTech/conv-bert-medium-small/resolve/main/config.json'
    ),
    'YituTech/conv-bert-small': 'https://huggingface.co/YituTech/conv-bert-small/resolve/main/config.json',
    # See all ConvBERT models at https://huggingface.co/models?filter=convbert
}
class  SCREAMING_SNAKE_CASE__  (				lowercase__			):
      snake_case__				:     Optional[int]        =						'''convbert'''
      def __init__(       self				:       List[Any]     ,							SCREAMING_SNAKE_CASE__				:       Optional[int]=3_0_5_2_2     ,							SCREAMING_SNAKE_CASE__				:       Dict=7_6_8     ,							SCREAMING_SNAKE_CASE__				:       Optional[int]=1_2     ,							SCREAMING_SNAKE_CASE__				:       Union[str, Any]=1_2     ,							SCREAMING_SNAKE_CASE__				:       str=3_0_7_2     ,							SCREAMING_SNAKE_CASE__				:       Dict="gelu"     ,							SCREAMING_SNAKE_CASE__				:       Dict=0.1     ,							SCREAMING_SNAKE_CASE__				:       Tuple=0.1     ,							SCREAMING_SNAKE_CASE__				:       List[str]=5_1_2     ,							SCREAMING_SNAKE_CASE__				:       Optional[Any]=2     ,							SCREAMING_SNAKE_CASE__				:       List[Any]=0.02     ,							SCREAMING_SNAKE_CASE__				:       Any=1E-12     ,							SCREAMING_SNAKE_CASE__				:       int=1     ,							SCREAMING_SNAKE_CASE__				:       int=0     ,							SCREAMING_SNAKE_CASE__				:       Optional[int]=2     ,							SCREAMING_SNAKE_CASE__				:       Optional[int]=7_6_8     ,							SCREAMING_SNAKE_CASE__				:       Union[str, Any]=2     ,							SCREAMING_SNAKE_CASE__				:       List[Any]=9     ,							SCREAMING_SNAKE_CASE__				:       List[Any]=1     ,							SCREAMING_SNAKE_CASE__				:       Dict=None     ,							**SCREAMING_SNAKE_CASE__				:       List[Any]     ,							)		->			Any:
             super().__init__(
                 pad_token_id=SCREAMING_SNAKE_CASE__     ,							bos_token_id=SCREAMING_SNAKE_CASE__     ,							eos_token_id=SCREAMING_SNAKE_CASE__     ,							**SCREAMING_SNAKE_CASE__     ,							)

             a_					:		Tuple							=  vocab_size
             a_					:		List[str]							=  hidden_size
             a_					:		List[str]							=  num_hidden_layers
             a_					:		Dict							=  num_attention_heads
             a_					:		Optional[int]							=  intermediate_size
             a_					:		int							=  hidden_act
             a_					:		Dict							=  hidden_dropout_prob
             a_					:		int							=  attention_probs_dropout_prob
             a_					:		str							=  max_position_embeddings
             a_					:		List[str]							=  type_vocab_size
             a_					:		List[str]							=  initializer_range
             a_					:		Tuple							=  layer_norm_eps
             a_					:		Optional[int]							=  embedding_size
             a_					:		List[Any]							=  head_ratio
             a_					:		List[Any]							=  conv_kernel_size
             a_					:		Tuple							=  num_groups
             a_					:		Tuple							=  classifier_dropout
class  SCREAMING_SNAKE_CASE__  (				lowercase__			):
      @property
      def        SCREAMING_SNAKE_CASE							(       self				:       List[str]     )		->			Mapping[str, Mapping[int, str]]:
             if self.task == "multiple-choice":
                    a_					:		Tuple							=  {0: 'batch', 1: 'choice', 2: 'sequence'}
             else:
                    a_					:		List[str]							=  {0: 'batch', 1: 'sequence'}
             return OrderedDict(
                 [
                     ('input_ids', dynamic_axis),
                     ('attention_mask', dynamic_axis),
                     ('token_type_ids', dynamic_axis),
                 ]     )


