
import gc
import unittest

import numpy as np
import torch

from diffusers import DDPMScheduler, MidiProcessor, SpectrogramDiffusionPipeline
from diffusers.pipelines.spectrogram_diffusion import SpectrogramContEncoder, SpectrogramNotesEncoder, TaFilmDecoder
from diffusers.utils import require_torch_gpu, skip_mps, slow, torch_device
from diffusers.utils.testing_utils import enable_full_determinism, require_note_seq, require_onnxruntime

from ..pipeline_params import TOKENS_TO_AUDIO_GENERATION_BATCH_PARAMS, TOKENS_TO_AUDIO_GENERATION_PARAMS
from ..test_pipelines_common import PipelineTesterMixin


enable_full_determinism()


a__				:      Optional[Any]    		=     '''./tests/fixtures/elise_format0.mid'''






@unittest.skip('The note-seq package currently throws an error on import'							)
class     a_       (			a__				,    unittest.TestCase							):



       """simple docstring"""
       __SCREAMING_SNAKE_CASE       :					Optional[Any]						=  SpectrogramDiffusionPipeline
       __SCREAMING_SNAKE_CASE       :					Dict						=  PipelineTesterMixin.required_optional_params - {
           'callback',
           'latents',
           'callback_steps',
           'output_type',
           'num_images_per_prompt',
       }
       __SCREAMING_SNAKE_CASE       :					Any						=  False
       __SCREAMING_SNAKE_CASE       :					Optional[Any]						=  False
       __SCREAMING_SNAKE_CASE       :					Tuple						=  TOKENS_TO_AUDIO_GENERATION_PARAMS
       __SCREAMING_SNAKE_CASE       :					List[Any]						=  TOKENS_TO_AUDIO_GENERATION_BATCH_PARAMS






       def 	__lowerCAmelCase     (		self   )					->List[str]:
           torch.manual_seed(0   )
           SCREAMING_SNAKE_CASE				:				Optional[Any]				      = SpectrogramNotesEncoder(
               max_length=2048				,       vocab_size=1536				,       d_model=768				,       dropout_rate=0.1				,       num_layers=1				,       num_heads=1				,       d_kv=4				,       d_ff=2048				,       feed_forward_proj='''gated-gelu'''				,       )

           SCREAMING_SNAKE_CASE				:				int				      = SpectrogramContEncoder(
               input_dims=128				,       targets_context_length=256				,       d_model=768				,       dropout_rate=0.1				,       num_layers=1				,       num_heads=1				,       d_kv=4				,       d_ff=2048				,       feed_forward_proj='''gated-gelu'''				,       )

           SCREAMING_SNAKE_CASE				:				int				      = TaFilmDecoder(
               input_dims=128				,       targets_length=256				,       max_decoder_noise_time=2_0_0_0_0.0				,       d_model=768				,       num_layers=1				,       num_heads=1				,       d_kv=4				,       d_ff=2048				,       dropout_rate=0.1				,       )

           SCREAMING_SNAKE_CASE				:				Dict				      = DDPMScheduler()

           SCREAMING_SNAKE_CASE				:				str				      = {
               '''notes_encoder''': notes_encoder.eval(),
               '''continuous_encoder''': continuous_encoder.eval(),
               '''decoder''': decoder.eval(),
               '''scheduler''': scheduler,
               '''melgan''': None,
           }
           return components






       def 	__lowerCAmelCase     (		self				,       _lowerCamelCase				,       _lowerCamelCase=0   )					->str:
           if str(_lowerCamelCase   ).startswith('''mps'''   ):
               SCREAMING_SNAKE_CASE				:				List[Any]				      = torch.manual_seed(_lowerCamelCase   )
           else:
               SCREAMING_SNAKE_CASE				:				Optional[int]				      = torch.Generator(device=_lowerCamelCase   ).manual_seed(_lowerCamelCase   )
           SCREAMING_SNAKE_CASE				:				int				      = {
               '''input_tokens''': [
                   [1134, 90, 1135, 1133, 1080, 112, 1132, 1080, 1133, 1079, 133, 1132, 1079, 1133, 1] + [0] * 2033
               ],
               '''generator''': generator,
               '''num_inference_steps''': 4,
               '''output_type''': '''mel''',
           }
           return inputs






       def 	__lowerCAmelCase     (		self   )					->Any:
           SCREAMING_SNAKE_CASE				:				Dict				      = '''cpu'''  # ensure determinism for the device-dependent torch.Generator
           SCREAMING_SNAKE_CASE				:				Any				      = self.get_dummy_components()
           SCREAMING_SNAKE_CASE				:				List[Any]				      = SpectrogramDiffusionPipeline(**_lowerCamelCase   )
           SCREAMING_SNAKE_CASE				:				Tuple				      = pipe.to(_lowerCamelCase   )
           pipe.set_progress_bar_config(disable=_lowerCamelCase   )

           SCREAMING_SNAKE_CASE				:				Optional[Any]				      = self.get_dummy_inputs(_lowerCamelCase   )
           SCREAMING_SNAKE_CASE				:				Union[str, Any]				      = pipe(**_lowerCamelCase   )
           SCREAMING_SNAKE_CASE				:				Union[str, Any]				      = output.audios

           SCREAMING_SNAKE_CASE				:				Optional[Any]				      = mel[0, -3:, -3:]

           assert mel_slice.shape == (3, 3)
           SCREAMING_SNAKE_CASE				:				Any				      = np.array(
               [-1_1.5_1_2_9_2_5, -4.7_8_8_2_1_5, -0.4_6_1_7_2_9_0_5, -2.0_5_1_7_1_5, -1_0.5_3_9_1_4_7, -1_0.9_7_0_9_6_3, -9.0_9_1_6_3_4, 4.0, 4.0]   )
           assert np.abs(mel_slice.flatten() - expected_slice   ).max() < 1e-2






       @skip_mps
       def 	__lowerCAmelCase     (		self   )					->Tuple:
           return super().test_save_load_local()






       @skip_mps
       def 	__lowerCAmelCase     (		self   )					->str:
           return super().test_dict_tuple_outputs_equivalent()






       @skip_mps
       def 	__lowerCAmelCase     (		self   )					->List[Any]:
           return super().test_save_load_optional_components()






       @skip_mps
       def 	__lowerCAmelCase     (		self   )					->Dict:
           return super().test_attention_slicing_forward_pass()






       def 	__lowerCAmelCase     (		self   )					->Tuple:
           pass






       def 	__lowerCAmelCase     (		self   )					->Tuple:
           pass






       @skip_mps
       def 	__lowerCAmelCase     (		self   )					->List[str]:
           return super().test_progress_bar()






@slow
@require_torch_gpu
@require_onnxruntime
@require_note_seq
class     a_       (			unittest.TestCase							):



       """simple docstring"""






       def 	__lowerCAmelCase     (		self   )					->int:
           # clean up the VRAM after each test
           super().tearDown()
           gc.collect()
           torch.cuda.empty_cache()






       def 	__lowerCAmelCase     (		self   )					->Optional[int]:
           # TODO - test that pipeline can decode tokens in a callback
           # so that music can be played live
           SCREAMING_SNAKE_CASE				:				str				      = torch_device

           SCREAMING_SNAKE_CASE				:				Tuple				      = SpectrogramDiffusionPipeline.from_pretrained('''google/music-spectrogram-diffusion'''   )
           SCREAMING_SNAKE_CASE				:				int				      = pipe.melgan
           SCREAMING_SNAKE_CASE				:				Any				      = None

           SCREAMING_SNAKE_CASE				:				List[Any]				      = pipe.to(_lowerCamelCase   )
           pipe.set_progress_bar_config(disable=_lowerCamelCase   )

           def callback(_lowerCamelCase				,       _lowerCamelCase   ):
               # decode mel to audio
               SCREAMING_SNAKE_CASE				:				Tuple				      = melgan(input_features=mel_output.astype(np.floataa   )   )[0]
               assert len(audio[0]   ) == 8_1920 * (step + 1)
               # simulate that audio is played
               return audio

           SCREAMING_SNAKE_CASE				:				List[Any]				      = MidiProcessor()
           SCREAMING_SNAKE_CASE				:				str				      = processor(_lowerCamelCase   )

           SCREAMING_SNAKE_CASE				:				Dict				      = input_tokens[:3]
           SCREAMING_SNAKE_CASE				:				Any				      = torch.manual_seed(0   )
           pipe(_lowerCamelCase				,       num_inference_steps=5				,       generator=_lowerCamelCase				,       callback=_lowerCamelCase				,       output_type='''mel'''   )






       def 	__lowerCAmelCase     (		self   )					->str:
           SCREAMING_SNAKE_CASE				:				Optional[Any]				      = torch_device

           SCREAMING_SNAKE_CASE				:				List[Any]				      = SpectrogramDiffusionPipeline.from_pretrained('''google/music-spectrogram-diffusion'''   )
           SCREAMING_SNAKE_CASE				:				Any				      = pipe.to(_lowerCamelCase   )
           pipe.set_progress_bar_config(disable=_lowerCamelCase   )
           SCREAMING_SNAKE_CASE				:				Optional[Any]				      = MidiProcessor()

           SCREAMING_SNAKE_CASE				:				Any				      = processor(_lowerCamelCase   )
           # just run two denoising loops
           SCREAMING_SNAKE_CASE				:				Tuple				      = input_tokens[:2]

           SCREAMING_SNAKE_CASE				:				Any				      = torch.manual_seed(0   )
           SCREAMING_SNAKE_CASE				:				Dict				      = pipe(_lowerCamelCase				,       num_inference_steps=2				,       generator=_lowerCamelCase   )

           SCREAMING_SNAKE_CASE				:				Dict				      = output.audios[0]

           assert abs(np.abs(_lowerCamelCase   ).sum() - 3_6_1_2.8_4_1   ) < 1e-1






       def 	__lowerCAmelCase     (		self   )					->Optional[Any]:
           SCREAMING_SNAKE_CASE				:				Tuple				      = torch_device

           SCREAMING_SNAKE_CASE				:				Tuple				      = SpectrogramDiffusionPipeline.from_pretrained('''google/music-spectrogram-diffusion'''   )
           SCREAMING_SNAKE_CASE				:				Optional[int]				      = pipe.to(_lowerCamelCase   )
           pipe.set_progress_bar_config(disable=_lowerCamelCase   )

           SCREAMING_SNAKE_CASE				:				int				      = MidiProcessor()

           SCREAMING_SNAKE_CASE				:				Optional[int]				      = processor(_lowerCamelCase   )

           # just run 4 denoising loops
           SCREAMING_SNAKE_CASE				:				Optional[Any]				      = input_tokens[:4]

           SCREAMING_SNAKE_CASE				:				Dict				      = torch.manual_seed(0   )
           SCREAMING_SNAKE_CASE				:				str				      = pipe(_lowerCamelCase				,       num_inference_steps=100				,       generator=_lowerCamelCase   )

           SCREAMING_SNAKE_CASE				:				Tuple				      = output.audios[0]
           assert abs(np.abs(_lowerCamelCase   ).sum() - 9_3_8_9.1_1_1_1   ) < 5e-2


from typing import TYPE_CHECKING

from ...utils import OptionalDependencyNotAvailable, _LazyModule, is_torch_available


a__				:      List[Any]    		=     {
    '''configuration_decision_transformer''': [
        '''DECISION_TRANSFORMER_PRETRAINED_CONFIG_ARCHIVE_MAP''',
        '''DecisionTransformerConfig''',
    ],
}

try:
      if not is_torch_available():
            raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
      pass
else:
      a__				:      List[str]    		=     [
          '''DECISION_TRANSFORMER_PRETRAINED_MODEL_ARCHIVE_LIST''',
          '''DecisionTransformerGPT2Model''',
          '''DecisionTransformerGPT2PreTrainedModel''',
          '''DecisionTransformerModel''',
          '''DecisionTransformerPreTrainedModel''',
      ]


if TYPE_CHECKING:
      from .configuration_decision_transformer import (
          DECISION_TRANSFORMER_PRETRAINED_CONFIG_ARCHIVE_MAP,
          DecisionTransformerConfig,
      )

      try:
            if not is_torch_available():
                  raise OptionalDependencyNotAvailable()
      except OptionalDependencyNotAvailable:
            pass
      else:
            from .modeling_decision_transformer import (
                DECISION_TRANSFORMER_PRETRAINED_MODEL_ARCHIVE_LIST,
                DecisionTransformerGPTaModel,
                DecisionTransformerGPTaPreTrainedModel,
                DecisionTransformerModel,
                DecisionTransformerPreTrainedModel,
            )


else:
      import sys

      a__				:      str    		=     _LazyModule(__name__, globals()['''__file__'''], _import_structure, module_spec=__spec__)
