






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

lowerCamelCase			:					List[str]         =	256



class 							_UpperCamelCase			(a_			):
      snake_case_  =							["""melgan"""]


      def __init__(	self					,						__UpperCamelCase					,						__UpperCamelCase					,						__UpperCamelCase					,						__UpperCamelCase					,						__UpperCamelCase					,						)->	None:
           super().__init__()

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

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


      def  __UpperCAmelCase		(	self					,						__UpperCamelCase					,						__UpperCamelCase=(-1.0, 1.0)					,						__UpperCamelCase=False	)->	List[str]:
           __lowerCAmelCase			, __lowerCAmelCase								=					output_range
           if clip:
                __lowerCAmelCase								=					torch.clip(__UpperCamelCase					,						self.min_value					,						self.max_value	)
           # Scale to [0, 1].
           __lowerCAmelCase								=					(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  __UpperCAmelCase		(	self					,						__UpperCamelCase					,						__UpperCamelCase=(-1.0, 1.0)					,						__UpperCamelCase=False	)->	int:
           __lowerCAmelCase			, __lowerCAmelCase								=					input_range
           __lowerCAmelCase								=					torch.clip(__UpperCamelCase					,						__UpperCamelCase					,						__UpperCamelCase	) if clip else outputs
           # Scale to [0, 1].
           __lowerCAmelCase								=					(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  __UpperCAmelCase		(	self					,						__UpperCamelCase					,						__UpperCamelCase					,						__UpperCamelCase	)->	Dict:
           __lowerCAmelCase								=					input_tokens > 0
           __lowerCAmelCase			, __lowerCAmelCase								=					self.notes_encoder(
               encoder_input_tokens=__UpperCamelCase					,						encoder_inputs_mask=__UpperCamelCase	)

           __lowerCAmelCase			, __lowerCAmelCase								=					self.continuous_encoder(
               encoder_inputs=__UpperCamelCase					,						encoder_inputs_mask=__UpperCamelCase	)

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


      def  __UpperCAmelCase		(	self					,						__UpperCamelCase					,						__UpperCamelCase					,						__UpperCamelCase	)->	str:
           __lowerCAmelCase								=					noise_time
           if not torch.is_tensor(__UpperCamelCase	):
                __lowerCAmelCase								=					torch.tensor([timesteps]					,						dtype=torch.long					,						device=input_tokens.device	)
           elif torch.is_tensor(__UpperCamelCase	) and len(timesteps.shape	) == 0:
                __lowerCAmelCase								=					timesteps[None].to(input_tokens.device	)

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

           __lowerCAmelCase								=					self.decoder(
               encodings_and_masks=__UpperCamelCase					,						decoder_input_tokens=__UpperCamelCase					,						decoder_noise_time=__UpperCamelCase	)
           return logits


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

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

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

                __lowerCAmelCase								=					self.scale_features(
                    __UpperCamelCase					,						output_range=[-1.0, 1.0]					,						clip=__UpperCamelCase	)

                __lowerCAmelCase								=					self.encode(
                    input_tokens=torch.IntTensor([encoder_input_tokens]	).to(device=self.device	)					,						continuous_inputs=__UpperCamelCase					,						continuous_mask=__UpperCamelCase					,						)

                # Sample encoder_continuous_inputs shaped gaussian noise to begin loop
                __lowerCAmelCase								=					randn_tensor(
                    shape=encoder_continuous_inputs.shape					,						generator=__UpperCamelCase					,						device=self.device					,						dtype=self.decoder.dtype					,						)

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

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

                     # Compute previous output: x_t -> x_t-1
                     __lowerCAmelCase								=					self.scheduler.step(__UpperCamelCase					,						__UpperCamelCase					,						__UpperCamelCase					,						generator=__UpperCamelCase	).prev_sample

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

                __lowerCAmelCase								=					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(__UpperCamelCase					,						__UpperCamelCase	)

                logger.info("Generated segment"					,						__UpperCamelCase	)

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

           if not return_dict:
                return (output,)

           return AudioPipelineOutput(audios=__UpperCamelCase	)














import os
import sys
import unittest


lowerCamelCase			:					int         =	os.path.abspath(os.path.dirname(os.path.dirname(os.path.dirname(__file__))))
sys.path.append(os.path.join(git_repo_path, '''utils'''))

import check_dummies  # noqa: E402
from check_dummies import create_dummy_files, create_dummy_object, find_backend, read_init  # noqa: E402


# Align TRANSFORMERS_PATH in check_dummies with the current path
lowerCamelCase			:					Any         =	os.path.join(git_repo_path, '''src''', '''diffusers''')



class 							_UpperCamelCase			(unittest.TestCase			):


      def  __UpperCAmelCase		(	self	)->	Optional[int]:
           __lowerCAmelCase								=					find_backend("    if not is_torch_available():"	)
           self.assertEqual(__UpperCamelCase					,						"torch"	)

           # backend_with_underscore = find_backend("    if not is_tensorflow_text_available():")
           # self.assertEqual(backend_with_underscore, "tensorflow_text")

           __lowerCAmelCase								=					find_backend("    if not (is_torch_available() and is_transformers_available()):"	)
           self.assertEqual(__UpperCamelCase					,						"torch_and_transformers"	)

           # double_backend_with_underscore = find_backend(
           #    "    if not (is_sentencepiece_available() and is_tensorflow_text_available()):"
           # )
           # self.assertEqual(double_backend_with_underscore, "sentencepiece_and_tensorflow_text")

           __lowerCAmelCase								=					find_backend(
               "    if not (is_torch_available() and is_transformers_available() and is_onnx_available()):"	)
           self.assertEqual(__UpperCamelCase					,						"torch_and_transformers_and_onnx"	)


      def  __UpperCAmelCase		(	self	)->	Dict:
           __lowerCAmelCase								=					read_init()
           # We don't assert on the exact list of keys to allow for smooth grow of backend-specific objects
           self.assertIn("torch"					,						__UpperCamelCase	)
           self.assertIn("torch_and_transformers"					,						__UpperCamelCase	)
           self.assertIn("flax_and_transformers"					,						__UpperCamelCase	)
           self.assertIn("torch_and_transformers_and_onnx"					,						__UpperCamelCase	)

           # Likewise, we can't assert on the exact content of a key
           self.assertIn("UNet2DModel"					,						objects["torch"]	)
           self.assertIn("FlaxUNet2DConditionModel"					,						objects["flax"]	)
           self.assertIn("StableDiffusionPipeline"					,						objects["torch_and_transformers"]	)
           self.assertIn("FlaxStableDiffusionPipeline"					,						objects["flax_and_transformers"]	)
           self.assertIn("LMSDiscreteScheduler"					,						objects["torch_and_scipy"]	)
           self.assertIn("OnnxStableDiffusionPipeline"					,						objects["torch_and_transformers_and_onnx"]	)


      def  __UpperCAmelCase		(	self	)->	Any:
           __lowerCAmelCase								=					create_dummy_object("CONSTANT"					,						"'torch'"	)
           self.assertEqual(__UpperCamelCase					,						"\nCONSTANT = None\n"	)

           __lowerCAmelCase								=					create_dummy_object("function"					,						"'torch'"	)
           self.assertEqual(
               __UpperCamelCase					,						"\ndef function(*args, **kwargs):\n    requires_backends(function, 'torch')\n"	)

           __lowerCAmelCase								=					"\nclass FakeClass(metaclass=DummyObject):\n    _backends = 'torch'\n\n    def __init__(self, *args, **kwargs):\n        requires_backends(self, 'torch')\n\n    @classmethod\n    def from_config(cls, *args, **kwargs):\n        requires_backends(cls, 'torch')\n\n    @classmethod\n    def from_pretrained(cls, *args, **kwargs):\n        requires_backends(cls, 'torch')\n"
           __lowerCAmelCase								=					create_dummy_object("FakeClass"					,						"'torch'"	)
           self.assertEqual(__UpperCamelCase					,						__UpperCamelCase	)


      def  __UpperCAmelCase		(	self	)->	Tuple:
           __lowerCAmelCase								=					"# This file is autogenerated by the command `make fix-copies`, do not edit.\nfrom ..utils import DummyObject, requires_backends\n\n\nCONSTANT = None\n\n\ndef function(*args, **kwargs):\n    requires_backends(function, [\"torch\"])\n\n\nclass FakeClass(metaclass=DummyObject):\n    _backends = [\"torch\"]\n\n    def __init__(self, *args, **kwargs):\n        requires_backends(self, [\"torch\"])\n\n    @classmethod\n    def from_config(cls, *args, **kwargs):\n        requires_backends(cls, [\"torch\"])\n\n    @classmethod\n    def from_pretrained(cls, *args, **kwargs):\n        requires_backends(cls, [\"torch\"])\n"
           __lowerCAmelCase								=					create_dummy_files({"torch": ["CONSTANT", "function", "FakeClass"]}	)
           self.assertEqual(dummy_files["torch"]					,						__UpperCamelCase	)






