




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			:	Dict							 =     logging.get_logger(__name__)  # pylint: disable=invalid-name

_lowerCamelCase			:	List[str]							 =     256


class        UpperCamelCase_					(				__lowerCAmelCase  ):

    '''simple docstring'''






    UpperCAmelCase__													=   ['''melgan''']


    def __init__(    self   :			Optional[int]      ,				UpperCAmelCase__   :			Optional[int]      ,				UpperCAmelCase__   :			List[Any]      ,				UpperCAmelCase__   :			Union[str, Any]      ,				UpperCAmelCase__   :			Optional[int]      ,				UpperCAmelCase__   :			Tuple      ,				)		->None:





           '''simple docstring'''

           super().__init__()

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

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


    def  SCREAMING_SNAKE_CASE							(    self   :			Dict      ,				UpperCAmelCase__   :			List[Any]      ,				UpperCAmelCase__   :			Optional[Any]=(-1.0, 1.0)      ,				UpperCAmelCase__   :			str=False)		->str:





           '''simple docstring'''

           A__				,							A__   				=					output_range
           if clip:
                  A__   				=					torch.clip(lowerCAmelCase_      ,				self.min_value      ,				self.max_value)
           # Scale to [0, 1].
           A__   				=					(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[int]      ,				UpperCAmelCase__   :			Tuple      ,				UpperCAmelCase__   :			Optional[int]=(-1.0, 1.0)      ,				UpperCAmelCase__   :			int=False)		->Optional[Any]:





           '''simple docstring'''

           A__				,							A__   				=					input_range
           A__   				=					torch.clip(lowerCAmelCase_      ,				lowerCAmelCase_      ,				lowerCAmelCase_) if clip else outputs
           # Scale to [0, 1].
           A__   				=					(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]      ,				UpperCAmelCase__   :			Optional[int]      ,				UpperCAmelCase__   :			Optional[Any]      ,				UpperCAmelCase__   :			Tuple)		->Optional[Any]:





           '''simple docstring'''

           A__   				=					input_tokens > 0
           A__				,							A__   				=					self.notes_encoder(
               encoder_input_tokens=lowerCAmelCase_      ,				encoder_inputs_mask=lowerCAmelCase_)

           A__				,							A__   				=					self.continuous_encoder(
               encoder_inputs=lowerCAmelCase_      ,				encoder_inputs_mask=lowerCAmelCase_)

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


    def  SCREAMING_SNAKE_CASE							(    self   :			Dict      ,				UpperCAmelCase__   :			Any      ,				UpperCAmelCase__   :			Optional[Any]      ,				UpperCAmelCase__   :			List[Any])		->int:





           '''simple docstring'''

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

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

           A__   				=					self.decoder(
               encodings_and_masks=lowerCAmelCase_      ,				decoder_input_tokens=lowerCAmelCase_      ,				decoder_noise_time=lowerCAmelCase_)
           return logits





    @torch.no_grad()
    def __call__(    self   :			Optional[int]      ,				UpperCAmelCase__   :			Any      ,				UpperCAmelCase__   :			Tuple = None      ,				UpperCAmelCase__   :			Optional[int] = 100      ,				UpperCAmelCase__   :			Optional[Any] = True      ,				UpperCAmelCase__   :			int = "numpy"      ,				UpperCAmelCase__   :			Dict = None      ,				UpperCAmelCase__   :			int = 1      ,				)		->Union[AudioPipelineOutput, Tuple]:





           '''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_)}.""")

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

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

                  A__   				=					self.scale_features(
                      lowerCAmelCase_      ,				output_range=[-1.0, 1.0]      ,				clip=lowerCAmelCase_)

                  A__   				=					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
                  A__   				=					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)):
                         A__   				=					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
                         A__   				=					self.scheduler.step(lowerCAmelCase_      ,				lowerCAmelCase_      ,				lowerCAmelCase_      ,				generator=lowerCAmelCase_).prev_sample

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

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

           if not return_dict:
                  return (output,)

           return AudioPipelineOutput(audios=lowerCAmelCase_)








from typing import List, Optional, Union

import numpy as np
import PIL.Image

from ...image_processing_utils import BaseImageProcessor, BatchFeature
from ...image_transforms import rescale, resize, to_channel_dimension_format
from ...image_utils import (
    ChannelDimension,
    PILImageResampling,
    get_image_size,
    make_list_of_images,
    to_numpy_array,
    valid_images,
)
from ...utils import TensorType, logging


_lowerCamelCase			:	str							 =     logging.get_logger(__name__)


class        UpperCamelCase_					(				UpperCAmelCase__  ):

    '''simple docstring'''






    UpperCAmelCase__													=   ['''pixel_values''']


    def __init__(    self   :			List[Any]      ,				UpperCAmelCase__   :			bool = True      ,				UpperCAmelCase__   :			int = 32      ,				UpperCAmelCase__   :			Dict=PILImageResampling.BILINEAR      ,				UpperCAmelCase__   :			bool = True      ,				**UpperCAmelCase__   :			Optional[int]      ,				)		->None:





           '''simple docstring'''

           A__   				=					do_resize
           A__   				=					do_rescale
           A__   				=					size_divisor
           A__   				=					resample
           super().__init__(**UpperCAmelCase__)


    def  SCREAMING_SNAKE_CASE							(    self   :			int      ,				UpperCAmelCase__   :			np.ndarray      ,				UpperCAmelCase__   :			int      ,				UpperCAmelCase__   :			List[str]      ,				UpperCAmelCase__   :			Optional[ChannelDimension] = None      ,				**UpperCAmelCase__   :			Dict)		->np.ndarray:





           '''simple docstring'''

           A__				,							A__   				=					get_image_size(UpperCAmelCase__)
           # Rounds the height and width down to the closest multiple of size_divisor
           A__   				=					height // size_divisor * size_divisor
           A__   				=					width // size_divisor * size_divisor
           A__   				=					resize(UpperCAmelCase__      ,				(new_h, new_w)      ,				resample=UpperCAmelCase__      ,				data_format=UpperCAmelCase__      ,				**UpperCAmelCase__)
           return image


    def  SCREAMING_SNAKE_CASE							(    self   :			List[Any]      ,				UpperCAmelCase__   :			np.ndarray      ,				UpperCAmelCase__   :			float      ,				UpperCAmelCase__   :			Optional[ChannelDimension] = None      ,				**UpperCAmelCase__   :			List[str])		->np.ndarray:





           '''simple docstring'''

           return rescale(image=UpperCAmelCase__      ,				scale=UpperCAmelCase__      ,				data_format=UpperCAmelCase__      ,				**UpperCAmelCase__)





    def  SCREAMING_SNAKE_CASE							(    self   :			Optional[int]      ,				UpperCAmelCase__   :			Union["PIL.Image.Image", TensorType, List["PIL.Image.Image"], List[TensorType]]      ,				UpperCAmelCase__   :			Optional[bool] = None      ,				UpperCAmelCase__   :			Optional[int] = None      ,				UpperCAmelCase__   :			Dict=None      ,				UpperCAmelCase__   :			Optional[bool] = None      ,				UpperCAmelCase__   :			Optional[Union[TensorType, str]] = None      ,				UpperCAmelCase__   :			ChannelDimension = ChannelDimension.FIRST      ,				**UpperCAmelCase__   :			List[str]      ,				)		->BatchFeature:





           '''simple docstring'''

           A__   				=					do_resize if do_resize is not None else self.do_resize
           A__   				=					do_rescale if do_rescale is not None else self.do_rescale
           A__   				=					size_divisor if size_divisor is not None else self.size_divisor
           A__   				=					resample if resample is not None else self.resample

           if do_resize and size_divisor is None:
                  raise ValueError('''size_divisor is required for resizing''')

           A__   				=					make_list_of_images(UpperCAmelCase__)

           if not valid_images(UpperCAmelCase__):
                  raise ValueError('''Invalid image(s)''')

           # All transformations expect numpy arrays.
           A__   				=					[to_numpy_array(UpperCAmelCase__) for img in images]

           if do_resize:
                  A__   				=					[self.resize(UpperCAmelCase__      ,				size_divisor=UpperCAmelCase__      ,				resample=UpperCAmelCase__) for image in images]

           if do_rescale:
                  A__   				=					[self.rescale(UpperCAmelCase__      ,				scale=1 / 255) for image in images]

           A__   				=					[to_channel_dimension_format(UpperCAmelCase__      ,				UpperCAmelCase__) for image in images]

           A__   				=					{'''pixel_values''': images}
           return BatchFeature(data=UpperCAmelCase__      ,				tensor_type=UpperCAmelCase__)


