




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()


UpperCamelCase__              =						'./tests/fixtures/elise_format0.mid'


@unittest.skip("""The note-seq package currently throws an error on import"""  )
class 				__SCREAMING_SNAKE_CASE							(		lowerCamelCase   ,     unittest.TestCase  ):
					_lowerCamelCase		  = SpectrogramDiffusionPipeline
					_lowerCamelCase		  = PipelineTesterMixin.required_optional_params - {
					    """callback""",
					    """latents""",
					    """callback_steps""",
					    """output_type""",
					    """num_images_per_prompt""",
					}
					_lowerCamelCase		  = False
					_lowerCamelCase		  = False
					_lowerCamelCase		  = TOKENS_TO_AUDIO_GENERATION_PARAMS
					_lowerCamelCase		  = TOKENS_TO_AUDIO_GENERATION_BATCH_PARAMS

					def 	lowercase_							(				self					):







						'''simple docstring'''




						torch.manual_seed(0					)
						lowerCamelCase_							:		List[Any]							   =							SpectrogramNotesEncoder(
						    max_length=2_048 ,       vocab_size=1_536 ,       d_model=768 ,       dropout_rate=0.1 ,       num_layers=1 ,       num_heads=1 ,       d_kv=4 ,       d_ff=2_048 ,       feed_forward_proj='gated-gelu' ,       )

						lowerCamelCase_							:		List[Any]							   =							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=2_048 ,       feed_forward_proj='gated-gelu' ,       )

						lowerCamelCase_							:		str							   =							TaFilmDecoder(
						    input_dims=128 ,       targets_length=256 ,       max_decoder_noise_time=2_00_00.0 ,       d_model=768 ,       num_layers=1 ,       num_heads=1 ,       d_kv=4 ,       d_ff=2_048 ,       dropout_rate=0.1 ,       )

						lowerCamelCase_							:		str							   =							DDPMScheduler()

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

					def 	lowercase_							(				self ,       __lowercase ,       __lowercase=0					):







						'''simple docstring'''




						if str(__lowercase					).startswith('mps'					):
							lowerCamelCase_							:		List[str]							   =							torch.manual_seed(__lowercase					)
						else:
							lowerCamelCase_							:		Union[str, Any]							   =							torch.Generator(device=__lowercase					).manual_seed(__lowercase					)
						lowerCamelCase_							:		Dict							   =							{
						    'input_tokens': [
						        [1_134, 90, 1_135, 1_133, 1_080, 112, 1_132, 1_080, 1_133, 1_079, 133, 1_132, 1_079, 1_133, 1] + [0] * 2_033
						    ],
						    'generator': generator,
						    'num_inference_steps': 4,
						    'output_type': 'mel',
						}
						return inputs

					def 	lowercase_							(				self					):







						'''simple docstring'''




						lowerCamelCase_							:		Tuple							   =							'cpu'  # ensure determinism for the device-dependent torch.Generator
						lowerCamelCase_							:		str							   =							self.get_dummy_components()
						lowerCamelCase_							:		Tuple							   =							SpectrogramDiffusionPipeline(**__lowercase					)
						lowerCamelCase_							:		List[str]							   =							pipe.to(__lowercase					)
						pipe.set_progress_bar_config(disable=__lowercase					)

						lowerCamelCase_							:		int							   =							self.get_dummy_inputs(__lowercase					)
						lowerCamelCase_							:		Optional[int]							   =							pipe(**__lowercase					)
						lowerCamelCase_							:		Any							   =							output.audios

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

						assert mel_slice.shape == (3, 3)
						lowerCamelCase_							:		Optional[Any]							   =							np.array(
						    [-11.51_29_25, -4.78_82_15, -0.46_17_29_05, -2.05_17_15, -10.53_91_47, -10.97_09_63, -9.09_16_34, 4.0, 4.0]					)
						assert np.abs(mel_slice.flatten() - expected_slice					).max() < 1E-2

					@skip_mps
					def 	lowercase_							(				self					):







						'''simple docstring'''




						return super().test_save_load_local()

					@skip_mps
					def 	lowercase_							(				self					):







						'''simple docstring'''




						return super().test_dict_tuple_outputs_equivalent()

					@skip_mps
					def 	lowercase_							(				self					):







						'''simple docstring'''




						return super().test_save_load_optional_components()

					@skip_mps
					def 	lowercase_							(				self					):







						'''simple docstring'''




						return super().test_attention_slicing_forward_pass()

					def 	lowercase_							(				self					):







						'''simple docstring'''




						pass

					def 	lowercase_							(				self					):







						'''simple docstring'''




						pass





					@skip_mps
					def 	lowercase_							(				self					):







						'''simple docstring'''




						return super().test_progress_bar()


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

					def 	lowercase_							(				self					):







						'''simple docstring'''




						super().tearDown()
						gc.collect()
						torch.cuda.empty_cache()

					def 	lowercase_							(				self					):







						'''simple docstring'''




						lowerCamelCase_							:		str							   =							torch_device

						lowerCamelCase_							:		Optional[Any]							   =							SpectrogramDiffusionPipeline.from_pretrained('google/music-spectrogram-diffusion'					)
						lowerCamelCase_							:		Tuple							   =							pipe.melgan
						lowerCamelCase_							:		List[str]							   =							None

						lowerCamelCase_							:		List[Any]							   =							pipe.to(__lowercase					)
						pipe.set_progress_bar_config(disable=__lowercase					)

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

						lowerCamelCase_							:		Tuple							   =							MidiProcessor()
						lowerCamelCase_							:		List[str]							   =							processor(__lowercase					)

						lowerCamelCase_							:		Optional[Any]							   =							input_tokens[:3]
						lowerCamelCase_							:		str							   =							torch.manual_seed(0					)
						pipe(__lowercase ,       num_inference_steps=5 ,       generator=__lowercase ,       callback=__lowercase ,       output_type='mel'					)

					def 	lowercase_							(				self					):







						'''simple docstring'''




						lowerCamelCase_							:		Dict							   =							torch_device

						lowerCamelCase_							:		Tuple							   =							SpectrogramDiffusionPipeline.from_pretrained('google/music-spectrogram-diffusion'					)
						lowerCamelCase_							:		str							   =							pipe.to(__lowercase					)
						pipe.set_progress_bar_config(disable=__lowercase					)
						lowerCamelCase_							:		str							   =							MidiProcessor()

						lowerCamelCase_							:		List[str]							   =							processor(__lowercase					)
						# just run two denoising loops
						lowerCamelCase_							:		List[str]							   =							input_tokens[:2]

						lowerCamelCase_							:		Union[str, Any]							   =							torch.manual_seed(0					)
						lowerCamelCase_							:		Optional[int]							   =							pipe(__lowercase ,       num_inference_steps=2 ,       generator=__lowercase					)

						lowerCamelCase_							:		Union[str, Any]							   =							output.audios[0]

						assert abs(np.abs(__lowercase					).sum() - 36_12.8_41					) < 1E-1





					def 	lowercase_							(				self					):







						'''simple docstring'''




						lowerCamelCase_							:		Tuple							   =							torch_device

						lowerCamelCase_							:		str							   =							SpectrogramDiffusionPipeline.from_pretrained('google/music-spectrogram-diffusion'					)
						lowerCamelCase_							:		List[Any]							   =							pipe.to(__lowercase					)
						pipe.set_progress_bar_config(disable=__lowercase					)

						lowerCamelCase_							:		Optional[int]							   =							MidiProcessor()

						lowerCamelCase_							:		str							   =							processor(__lowercase					)

						# just run 4 denoising loops
						lowerCamelCase_							:		Dict							   =							input_tokens[:4]

						lowerCamelCase_							:		Tuple							   =							torch.manual_seed(0					)
						lowerCamelCase_							:		str							   =							pipe(__lowercase ,       num_inference_steps=100 ,       generator=__lowercase					)

						lowerCamelCase_							:		List[str]							   =							output.audios[0]
						assert abs(np.abs(__lowercase					).sum() - 93_89.11_11					) < 5E-2








"""simple docstring"""



import gym
import numpy as np
import torch
from mujoco_py import GlfwContext

from transformers import DecisionTransformerModel


GlfwContext(offscreen=True)  # Create a window to init GLFW.


def      SCREAMING_SNAKE_CASE__    (   snake_case    :      Any       , snake_case    :      Optional[int]       , snake_case    :      Tuple       , snake_case    :      Union[str, Any]       , snake_case    :      Dict       , snake_case    :      str							)->						Optional[Any]:






     '''simple docstring'''



     UpperCAmelCase__		:  List[Any]				=				states.reshape(1       , -1       , model.config.state_dim							)
     UpperCAmelCase__		:  str				=				actions.reshape(1       , -1       , model.config.act_dim							)
     UpperCAmelCase__		:  Any				=				returns_to_go.reshape(1       , -1       , 1							)
     UpperCAmelCase__		:  Union[str, Any]				=				timesteps.reshape(1       , -1							)

     if model.config.max_length is not None:
          UpperCAmelCase__		:  Tuple				=				states[:, -model.config.max_length :]
          UpperCAmelCase__		:  int				=				actions[:, -model.config.max_length :]
          UpperCAmelCase__		:  Union[str, Any]				=				returns_to_go[:, -model.config.max_length :]
          UpperCAmelCase__		:  int				=				timesteps[:, -model.config.max_length :]

          # pad all tokens to sequence length
          UpperCAmelCase__		:  Optional[Any]				=				torch.cat(
              [torch.zeros(model.config.max_length - states.shape[1]							), torch.ones(states.shape[1]							)]							)
          UpperCAmelCase__		:  Optional[Any]				=				attention_mask.to(dtype=torch.long       , device=states.device							).reshape(1       , -1							)
          UpperCAmelCase__		:  Optional[int]				=				torch.cat(
              [
                  torch.zeros(
                      (states.shape[0], model.config.max_length - states.shape[1], model.config.state_dim)       , device=states.device       , ),
                  states,
              ]       , dim=1       , ).to(dtype=torch.floataa							)
          UpperCAmelCase__		:  Optional[Any]				=				torch.cat(
              [
                  torch.zeros(
                      (actions.shape[0], model.config.max_length - actions.shape[1], model.config.act_dim)       , device=actions.device       , ),
                  actions,
              ]       , dim=1       , ).to(dtype=torch.floataa							)
          UpperCAmelCase__		:  Any				=				torch.cat(
              [
                  torch.zeros(
                      (returns_to_go.shape[0], model.config.max_length - returns_to_go.shape[1], 1)       , device=returns_to_go.device       , ),
                  returns_to_go,
              ]       , dim=1       , ).to(dtype=torch.floataa							)
          UpperCAmelCase__		:  str				=				torch.cat(
              [
                  torch.zeros(
                      (timesteps.shape[0], model.config.max_length - timesteps.shape[1])       , device=timesteps.device							),
                  timesteps,
              ]       , dim=1       , ).to(dtype=torch.long							)
     else:
          UpperCAmelCase__		:  Any				=				None

     UpperCAmelCase__				,							UpperCAmelCase__				,							UpperCAmelCase__		:  List[str]				=				model(
         states=snake_case       , actions=snake_case       , rewards=snake_case       , returns_to_go=snake_case       , timesteps=snake_case       , attention_mask=snake_case       , return_dict=snake_case       , )

     return action_preds[0, -1]


# build the environment

_lowerCAmelCase     :     Dict            =	gym.make("""Hopper-v3""")
_lowerCAmelCase     :     Any            =	env.observation_space.shape[0]
_lowerCAmelCase     :     List[Any]            =	env.action_space.shape[0]
_lowerCAmelCase     :     Any            =	1_000
_lowerCAmelCase     :     Union[str, Any]            =	"""cuda"""
_lowerCAmelCase     :     Optional[Any]            =	10_00.0  # normalization for rewards/returns
_lowerCAmelCase     :     Optional[int]            =	3_600 / scale  # evaluation conditioning targets, 3600 is reasonable from the paper LINK
_lowerCAmelCase     :     Any            =	np.array(
    [
        1.31_12_79,
        -0.08_46_95_21,
        -0.5_38_27_19,
        -0.07_20_15_76,
        0.04_93_23_66,
        2.1_06_68_56,
        -0.15_01_73_54,
        0.00_87_83_45,
        -0.2_84_81_86,
        -0.18_54_00_96,
        -0.28_46_12_86,
    ]
)
_lowerCAmelCase     :     Optional[Any]            =	np.array(
    [
        0.17_79_07_51,
        0.05_44_46_21,
        0.21_29_71_39,
        0.14_53_04_19,
        0.6_12_44_44,
        0.85_17_44_65,
        1.4_51_52_52,
        0.6_75_16_96,
        1.53_62_39,
        1.6_16_07_46,
        5.6_07_25_36,
    ]
)
_lowerCAmelCase     :     Any            =	torch.from_numpy(state_mean).to(device=device)
_lowerCAmelCase     :     List[Any]            =	torch.from_numpy(state_std).to(device=device)

# Create the decision transformer model
_lowerCAmelCase     :     List[str]            =	DecisionTransformerModel.from_pretrained("""edbeeching/decision-transformer-gym-hopper-medium""")
_lowerCAmelCase     :     Optional[int]            =	model.to(device)
model.eval()

for ep in range(10):
 _lowerCAmelCase       ,   _lowerCAmelCase     :     Dict            =	0, 0
 _lowerCAmelCase     :     Dict            =	env.reset()
 _lowerCAmelCase     :     str            =	torch.tensor(TARGET_RETURN, device=device, dtype=torch.floataa).reshape(1, 1)
 _lowerCAmelCase     :     Union[str, Any]            =	torch.from_numpy(state).reshape(1, state_dim).to(device=device, dtype=torch.floataa)
 _lowerCAmelCase     :     List[Any]            =	torch.zeros((0, act_dim), device=device, dtype=torch.floataa)
 _lowerCAmelCase     :     Dict            =	torch.zeros(0, device=device, dtype=torch.floataa)

 _lowerCAmelCase     :     Optional[int]            =	torch.tensor(0, device=device, dtype=torch.long).reshape(1, 1)
 for t in range(max_ep_len):
  env.render()
  # add padding
  _lowerCAmelCase     :     int            =	torch.cat([actions, torch.zeros((1, act_dim), device=device)], dim=0)
  _lowerCAmelCase     :     str            =	torch.cat([rewards, torch.zeros(1, device=device)])

  _lowerCAmelCase     :     Tuple            =	get_action(
      model,
      (states.to(dtype=torch.floataa) - state_mean) / state_std,
      actions.to(dtype=torch.floataa),
      rewards.to(dtype=torch.floataa),
      target_return.to(dtype=torch.floataa),
      timesteps.to(dtype=torch.long),
  )
  _lowerCAmelCase     :     List[str]            =	action
  _lowerCAmelCase     :     Optional[int]            =	action.detach().cpu().numpy()

  _lowerCAmelCase       ,   _lowerCAmelCase       ,   _lowerCAmelCase       ,   _lowerCAmelCase     :     int            =	env.step(action)

  _lowerCAmelCase     :     Union[str, Any]            =	torch.from_numpy(state).to(device=device).reshape(1, state_dim)
  _lowerCAmelCase     :     Optional[int]            =	torch.cat([states, cur_state], dim=0)
  _lowerCAmelCase     :     List[Any]            =	reward

  _lowerCAmelCase     :     List[str]            =	target_return[0, -1] - (reward / scale)
  _lowerCAmelCase     :     Optional[int]            =	torch.cat([target_return, pred_return.reshape(1, 1)], dim=1)
  _lowerCAmelCase     :     Tuple            =	torch.cat([timesteps, torch.ones((1, 1), device=device, dtype=torch.long) * (t + 1)], dim=1)

  episode_return += reward
  episode_length += 1

  if done:
   break


