




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


lowerCamelCase__				:     List[Any]			       =							'./tests/fixtures/elise_format0.mid'





@unittest.skip("The note-seq package currently throws an error on import"      )
class  lowerCamelCase_							(   _SCREAMING_SNAKE_CASE ,     unittest.TestCase      ):


 '''simple docstring'''







 lowercase_       					=					SpectrogramDiffusionPipeline
 lowercase_       					=					PipelineTesterMixin.required_optional_params - {
     "callback",
     "latents",
     "callback_steps",
     "output_type",
     "num_images_per_prompt",
 }
 lowercase_       					=					False
 lowercase_       					=					False
 lowercase_       					=					TOKENS_TO_AUDIO_GENERATION_PARAMS
 lowercase_       					=					TOKENS_TO_AUDIO_GENERATION_BATCH_PARAMS



 def    lowerCAmelCase_						(							self						:	Any		):
       torch.manual_seed(0		)
       SCREAMING_SNAKE_CASE_					=    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'       ,       )

       SCREAMING_SNAKE_CASE_					=    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'       ,       )

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

       SCREAMING_SNAKE_CASE_					=    DDPMScheduler()

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



 def    lowerCAmelCase_						(							self						:	Tuple       ,       _lowerCAmelCase						:	int       ,       _lowerCAmelCase						:	Optional[Any]=0		):
       if str(_lowerCAmelCase		).startswith('mps'		):
             SCREAMING_SNAKE_CASE_					=    torch.manual_seed(_lowerCAmelCase		)
       else:
             SCREAMING_SNAKE_CASE_					=    torch.Generator(device=_lowerCAmelCase		).manual_seed(_lowerCAmelCase		)
       SCREAMING_SNAKE_CASE_					=    {
           '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    lowerCAmelCase_						(							self						:	Dict		):
       SCREAMING_SNAKE_CASE_					=    'cpu'  # ensure determinism for the device-dependent torch.Generator
       SCREAMING_SNAKE_CASE_					=    self.get_dummy_components()
       SCREAMING_SNAKE_CASE_					=    SpectrogramDiffusionPipeline(**_lowerCAmelCase		)
       SCREAMING_SNAKE_CASE_					=    pipe.to(_lowerCAmelCase		)
       pipe.set_progress_bar_config(disable=_lowerCAmelCase		)

       SCREAMING_SNAKE_CASE_					=    self.get_dummy_inputs(_lowerCAmelCase		)
       SCREAMING_SNAKE_CASE_					=    pipe(**_lowerCAmelCase		)
       SCREAMING_SNAKE_CASE_					=    output.audios

       SCREAMING_SNAKE_CASE_					=    mel[0, -3:, -3:]

       assert mel_slice.shape == (3, 3)
       SCREAMING_SNAKE_CASE_					=    np.array(
           [-11.51_2925, -4.78_8215, -0.4617_2905, -2.05_1715, -10.53_9147, -10.97_0963, -9.09_1634, 4.0, 4.0]		)
       assert np.abs(mel_slice.flatten() - expected_slice		).max() < 1E-2



 @skip_mps
 def    lowerCAmelCase_						(							self						:	str		):
       return super().test_save_load_local()



 @skip_mps
 def    lowerCAmelCase_						(							self						:	Tuple		):
       return super().test_dict_tuple_outputs_equivalent()



 @skip_mps
 def    lowerCAmelCase_						(							self						:	Union[str, Any]		):
       return super().test_save_load_optional_components()



 @skip_mps
 def    lowerCAmelCase_						(							self						:	List[str]		):
       return super().test_attention_slicing_forward_pass()



 def    lowerCAmelCase_						(							self						:	Dict		):
       pass



 def    lowerCAmelCase_						(							self						:	Tuple		):
       pass




 @skip_mps
 def    lowerCAmelCase_						(							self						:	Optional[int]		):
       return super().test_progress_bar()





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


 '''simple docstring'''







 def    lowerCAmelCase_						(							self						:	List[Any]		):
       # clean up the VRAM after each test
       super().tearDown()
       gc.collect()
       torch.cuda.empty_cache()



 def    lowerCAmelCase_						(							self						:	Optional[Any]		):
       # TODO - test that pipeline can decode tokens in a callback
       # so that music can be played live
       SCREAMING_SNAKE_CASE_					=    torch_device

       SCREAMING_SNAKE_CASE_					=    SpectrogramDiffusionPipeline.from_pretrained('google/music-spectrogram-diffusion'		)
       SCREAMING_SNAKE_CASE_					=    pipe.melgan
       SCREAMING_SNAKE_CASE_					=    None

       SCREAMING_SNAKE_CASE_					=    pipe.to(_lowerCAmelCase		)
       pipe.set_progress_bar_config(disable=_lowerCAmelCase		)

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

       SCREAMING_SNAKE_CASE_					=    MidiProcessor()
       SCREAMING_SNAKE_CASE_					=    processor(_lowerCAmelCase		)

       SCREAMING_SNAKE_CASE_					=    input_tokens[:3]
       SCREAMING_SNAKE_CASE_					=    torch.manual_seed(0		)
       pipe(_lowerCAmelCase       ,       num_inference_steps=5       ,       generator=_lowerCAmelCase       ,       callback=_lowerCAmelCase       ,       output_type='mel'		)



 def    lowerCAmelCase_						(							self						:	Union[str, Any]		):
       SCREAMING_SNAKE_CASE_					=    torch_device

       SCREAMING_SNAKE_CASE_					=    SpectrogramDiffusionPipeline.from_pretrained('google/music-spectrogram-diffusion'		)
       SCREAMING_SNAKE_CASE_					=    pipe.to(_lowerCAmelCase		)
       pipe.set_progress_bar_config(disable=_lowerCAmelCase		)
       SCREAMING_SNAKE_CASE_					=    MidiProcessor()

       SCREAMING_SNAKE_CASE_					=    processor(_lowerCAmelCase		)
       # just run two denoising loops
       SCREAMING_SNAKE_CASE_					=    input_tokens[:2]

       SCREAMING_SNAKE_CASE_					=    torch.manual_seed(0		)
       SCREAMING_SNAKE_CASE_					=    pipe(_lowerCAmelCase       ,       num_inference_steps=2       ,       generator=_lowerCAmelCase		)

       SCREAMING_SNAKE_CASE_					=    output.audios[0]

       assert abs(np.abs(_lowerCAmelCase		).sum() - 3612.841		) < 1E-1




 def    lowerCAmelCase_						(							self						:	Optional[int]		):
       SCREAMING_SNAKE_CASE_					=    torch_device

       SCREAMING_SNAKE_CASE_					=    SpectrogramDiffusionPipeline.from_pretrained('google/music-spectrogram-diffusion'		)
       SCREAMING_SNAKE_CASE_					=    pipe.to(_lowerCAmelCase		)
       pipe.set_progress_bar_config(disable=_lowerCAmelCase		)

       SCREAMING_SNAKE_CASE_					=    MidiProcessor()

       SCREAMING_SNAKE_CASE_					=    processor(_lowerCAmelCase		)

       # just run 4 denoising loops
       SCREAMING_SNAKE_CASE_					=    input_tokens[:4]

       SCREAMING_SNAKE_CASE_					=    torch.manual_seed(0		)
       SCREAMING_SNAKE_CASE_					=    pipe(_lowerCAmelCase       ,       num_inference_steps=100       ,       generator=_lowerCAmelCase		)

       SCREAMING_SNAKE_CASE_					=    output.audios[0]
       assert abs(np.abs(_lowerCAmelCase		).sum() - 9389.1111		) < 5E-2





import os
from pathlib import Path
from typing import Dict, List

import fire
import torch

from transformers import AutoModelForSeqaSeqLM, AutoTokenizer
from transformers.utils.logging import get_logger


lowerCamelCase__				:     Optional[Any]			       =							get_logger(__name__)

def       UpperCAmelCase_							(   __UpperCAmelCase		:			str    ,    __UpperCAmelCase		:			str							)      ->				List[str]:
      if text.startswith(__UpperCAmelCase							):
            return text[len(__UpperCAmelCase							) :]
      return text  # or whatever

def       UpperCAmelCase_							(   __UpperCAmelCase		:			Union[str, Any]							)      ->				List[Any]:
      return {remove_prefix(__UpperCAmelCase    ,    'model.'							): v for k, v in sd.items()}

def       UpperCAmelCase_							(   __UpperCAmelCase		:			List[Dict[str, torch.Tensor]]							)      ->				Dict:
      SCREAMING_SNAKE_CASE_					=    {}
      for k in state_dicts[0].keys():
            SCREAMING_SNAKE_CASE_					=    [sd[k] for sd in state_dicts]
            SCREAMING_SNAKE_CASE_					=    sum(__UpperCAmelCase							) / len(__UpperCAmelCase							)
            assert isinstance(__UpperCAmelCase    ,    torch.Tensor							)
            SCREAMING_SNAKE_CASE_					=    new_t
      return new_sd

def       UpperCAmelCase_							(   __UpperCAmelCase		:			str    ,    __UpperCAmelCase		:			str    ,    __UpperCAmelCase		:			str							)      ->				None:
      SCREAMING_SNAKE_CASE_					=    AutoModelForSeqaSeqLM.from_pretrained(__UpperCAmelCase							)
      if os.path.isfile(__UpperCAmelCase							):
            SCREAMING_SNAKE_CASE_					=    [pl_ckpt_path]
      else:
            assert os.path.isdir(__UpperCAmelCase							)
            SCREAMING_SNAKE_CASE_					=    list(Path(__UpperCAmelCase							).glob('*.ckpt'							)							)
            assert ckpt_files, f"could not find any ckpt files inside the {pl_ckpt_path} directory"

      if len(__UpperCAmelCase							) > 1:
            logger.info(f"averaging the weights of {ckpt_files}"							)

      SCREAMING_SNAKE_CASE_					=    [sanitize(torch.load(__UpperCAmelCase    ,    map_location='cpu'							)['state_dict']							) for x in ckpt_files]
      SCREAMING_SNAKE_CASE_					=    average_state_dicts(__UpperCAmelCase							)

      SCREAMING_SNAKE_CASE_		,				SCREAMING_SNAKE_CASE_					=    hf_model.load_state_dict(__UpperCAmelCase    ,    strict=__UpperCAmelCase							)
      assert not missing, f"missing keys: {missing}"
      hf_model.save_pretrained(__UpperCAmelCase							)
      try:
            SCREAMING_SNAKE_CASE_					=    AutoTokenizer.from_pretrained(__UpperCAmelCase							)
            tok.save_pretrained(__UpperCAmelCase							)
      except Exception:
            pass
            # dont copy tokenizer if cant


if __name__ == "__main__":
   fire.Fire(convert_pl_to_hf)