




'''simple docstring'''

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__         =     '''./tests/fixtures/elise_format0.mid'''





@unittest.skip("""The note-seq package currently throws an error on import"""      )
class __magic_name__(						__lowerCAmelCase    ,		unittest.TestCase      ):
						UpperCAmelCase_		:					Tuple    =  SpectrogramDiffusionPipeline
						UpperCAmelCase_		:					List[str]    =  PipelineTesterMixin.required_optional_params - {
						    """callback""",
						    """latents""",
						    """callback_steps""",
						    """output_type""",
						    """num_images_per_prompt""",
						}
						UpperCAmelCase_		:					Dict    =  False
						UpperCAmelCase_		:					Dict    =  False
						UpperCAmelCase_		:					List[str]    =  TOKENS_TO_AUDIO_GENERATION_PARAMS
						UpperCAmelCase_		:					Union[str, Any]    =  TOKENS_TO_AUDIO_GENERATION_BATCH_PARAMS






						def       __lowerCAmelCase(   self						:						Tuple       ):



									'''simple docstring'''






									torch.manual_seed(0       )
									snake_case__          =						SpectrogramNotesEncoder(
									    max_length=2_0_4_8					,   vocab_size=1_5_3_6					,   d_model=7_6_8					,   dropout_rate=0.1					,   num_layers=1					,   num_heads=1					,   d_kv=4					,   d_ff=2_0_4_8					,   feed_forward_proj="""gated-gelu"""					,   )

									snake_case__          =						SpectrogramContEncoder(
									    input_dims=1_2_8					,   targets_context_length=2_5_6					,   d_model=7_6_8					,   dropout_rate=0.1					,   num_layers=1					,   num_heads=1					,   d_kv=4					,   d_ff=2_0_4_8					,   feed_forward_proj="""gated-gelu"""					,   )

									snake_case__          =						TaFilmDecoder(
									    input_dims=1_2_8					,   targets_length=2_5_6					,   max_decoder_noise_time=2_00_00.0					,   d_model=7_6_8					,   num_layers=1					,   num_heads=1					,   d_kv=4					,   d_ff=2_0_4_8					,   dropout_rate=0.1					,   )

									snake_case__          =						DDPMScheduler()

									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					,   __UpperCamelCase						:						Any					,   __UpperCamelCase						:						str=0       ):



									'''simple docstring'''






									if str(__UpperCamelCase       ).startswith("""mps"""       ):
												snake_case__          =						torch.manual_seed(__UpperCamelCase       )
									else:
												snake_case__          =						torch.Generator(device=__UpperCamelCase       ).manual_seed(__UpperCamelCase       )
									snake_case__          =						{
									    """input_tokens""": [
									        [1_1_3_4, 9_0, 1_1_3_5, 1_1_3_3, 1_0_8_0, 1_1_2, 1_1_3_2, 1_0_8_0, 1_1_3_3, 1_0_7_9, 1_3_3, 1_1_3_2, 1_0_7_9, 1_1_3_3, 1] + [0] * 2_0_3_3
									    ],
									    """generator""": generator,
									    """num_inference_steps""": 4,
									    """output_type""": """mel""",
									}
									return inputs






						def       __lowerCAmelCase(   self						:						Optional[Any]       ):



									'''simple docstring'''






									snake_case__          =						"""cpu"""  # ensure determinism for the device-dependent torch.Generator
									snake_case__          =						self.get_dummy_components()
									snake_case__          =						SpectrogramDiffusionPipeline(**__UpperCamelCase       )
									snake_case__          =						pipe.to(__UpperCamelCase       )
									pipe.set_progress_bar_config(disable=__UpperCamelCase       )

									snake_case__          =						self.get_dummy_inputs(__UpperCamelCase       )
									snake_case__          =						pipe(**__UpperCamelCase       )
									snake_case__          =						output.audios

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

									assert mel_slice.shape == (3, 3)
									snake_case__          =						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       __lowerCAmelCase(   self						:						Tuple       ):



									'''simple docstring'''






									return super().test_save_load_local()






						@skip_mps
						def       __lowerCAmelCase(   self						:						List[Any]       ):



									'''simple docstring'''






									return super().test_dict_tuple_outputs_equivalent()






						@skip_mps
						def       __lowerCAmelCase(   self						:						Optional[int]       ):



									'''simple docstring'''






									return super().test_save_load_optional_components()






						@skip_mps
						def       __lowerCAmelCase(   self						:						Union[str, Any]       ):



									'''simple docstring'''






									return super().test_attention_slicing_forward_pass()






						def       __lowerCAmelCase(   self						:						Dict       ):



									'''simple docstring'''






									pass






						def       __lowerCAmelCase(   self						:						int       ):



									'''simple docstring'''






									pass







						@skip_mps
						def       __lowerCAmelCase(   self						:						Dict       ):



									'''simple docstring'''






									return super().test_progress_bar()







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






						def       __lowerCAmelCase(   self						:						List[Any]       ):



									'''simple docstring'''






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






						def       __lowerCAmelCase(   self						:						int       ):



									'''simple docstring'''






									snake_case__          =						torch_device

									snake_case__          =						SpectrogramDiffusionPipeline.from_pretrained("""google/music-spectrogram-diffusion"""       )
									snake_case__          =						pipe.melgan
									snake_case__          =						None

									snake_case__          =						pipe.to(__UpperCamelCase       )
									pipe.set_progress_bar_config(disable=__UpperCamelCase       )

									def callback(__UpperCamelCase						:						List[str]					,   __UpperCamelCase						:						List[str]       ):
												# decode mel to audio
												snake_case__          =						melgan(input_features=mel_output.astype(np.floataa       )       )[0]
												assert len(audio[0]       ) == 8_1_9_2_0 * (step + 1)
												# simulate that audio is played
												return audio

									snake_case__          =						MidiProcessor()
									snake_case__          =						processor(__UpperCamelCase       )

									snake_case__          =						input_tokens[:3]
									snake_case__          =						torch.manual_seed(0       )
									pipe(__UpperCamelCase					,   num_inference_steps=5					,   generator=__UpperCamelCase					,   callback=__UpperCamelCase					,   output_type="""mel"""       )






						def       __lowerCAmelCase(   self						:						List[Any]       ):



									'''simple docstring'''






									snake_case__          =						torch_device

									snake_case__          =						SpectrogramDiffusionPipeline.from_pretrained("""google/music-spectrogram-diffusion"""       )
									snake_case__          =						pipe.to(__UpperCamelCase       )
									pipe.set_progress_bar_config(disable=__UpperCamelCase       )
									snake_case__          =						MidiProcessor()

									snake_case__          =						processor(__UpperCamelCase       )
									# just run two denoising loops
									snake_case__          =						input_tokens[:2]

									snake_case__          =						torch.manual_seed(0       )
									snake_case__          =						pipe(__UpperCamelCase					,   num_inference_steps=2					,   generator=__UpperCamelCase       )

									snake_case__          =						output.audios[0]

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







						def       __lowerCAmelCase(   self						:						Union[str, Any]       ):



									'''simple docstring'''






									snake_case__          =						torch_device

									snake_case__          =						SpectrogramDiffusionPipeline.from_pretrained("""google/music-spectrogram-diffusion"""       )
									snake_case__          =						pipe.to(__UpperCamelCase       )
									pipe.set_progress_bar_config(disable=__UpperCamelCase       )

									snake_case__          =						MidiProcessor()

									snake_case__          =						processor(__UpperCamelCase       )

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

									snake_case__          =						torch.manual_seed(0       )
									snake_case__          =						pipe(__UpperCamelCase					,   num_inference_steps=1_0_0					,   generator=__UpperCamelCase       )

									snake_case__          =						output.audios[0]
									assert abs(np.abs(__UpperCamelCase       ).sum() - 93_89.11_11       ) < 5E-2






import json
import os
from functools import lru_cache
from typing import List, Optional, Tuple

import regex as re

from ...tokenization_utils import AddedToken, PreTrainedTokenizer
from ...utils import logging


_UpperCamelCase       :				Optional[Any]           =			logging.get_logger(__name__)


_UpperCamelCase       :				Optional[int]           =			{"""vocab_file""": """vocab.json""", """merges_file""": """merges.txt"""}

# See all MVP models at https://huggingface.co/models?filter=mvp
_UpperCamelCase       :				int           =			{
    """vocab_file""": {
        """RUCAIBox/mvp""": """https://huggingface.co/RUCAIBox/mvp/resolve/main/vocab.json""",
    },
    """added_tokens.json""": {
        """RUCAIBox/mvp""": """https://huggingface.co/RUCAIBox/mvp/resolve/main/added_tokens.json""",
    },
    """merges_file""": {
        """RUCAIBox/mvp""": """https://huggingface.co/RUCAIBox/mvp/resolve/main/merges.txt""",
    },
}

_UpperCamelCase       :				str           =			{
    """RUCAIBox/mvp""": 1_0_2_4,
}
@lru_cache()
def  __UpperCamelCase    (						)    ->		Tuple:


   '''simple docstring'''







   __A    =  (
       list(range(ord('''!'''      )							,			ord('''~'''      ) + 1      )      ) + list(range(ord('''¡'''      )							,			ord('''¬'''      ) + 1      )      ) + list(range(ord('''®'''      )							,			ord('''ÿ'''      ) + 1      )      )
   )
   __A    =  bs[:]
   __A    =  0
   for b in range(2**8      ):
      if b not in bs:
         bs.append(snake_case      )
         cs.append(2**8 + n      )
         n += 1
   __A    =  [chr(snake_case      ) for n in cs]
   return dict(zip(snake_case							,			snake_case      )      )
def  __UpperCamelCase    (						snake_case      )    ->		int:


   '''simple docstring'''







   __A    =  set()
   __A    =  word[0]
   for char in word[1:]:
      pairs.add((prev_char, char)      )
      __A    =  char
   return pairs







class 		_lowerCAmelCase(		_a):

  """simple docstring"""







  lowerCamelCase__     	=       VOCAB_FILES_NAMES
  lowerCamelCase__     	=       PRETRAINED_VOCAB_FILES_MAP
  lowerCamelCase__     	=       PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
  lowerCamelCase__     	=       ['''input_ids''', '''attention_mask''']

  def __init__(   self    ,							UpperCAmelCase    ,							UpperCAmelCase    ,							UpperCAmelCase="replace"    ,							UpperCAmelCase="<s>"    ,							UpperCAmelCase="</s>"    ,							UpperCAmelCase="</s>"    ,							UpperCAmelCase="<s>"    ,							UpperCAmelCase="<unk>"    ,							UpperCAmelCase="<pad>"    ,							UpperCAmelCase="<mask>"    ,							UpperCAmelCase=False    ,							**UpperCAmelCase    ,							)->   Optional[int]:
     __A    =  AddedToken(UpperCAmelCase    ,							lstrip=UpperCAmelCase    ,							rstrip=UpperCAmelCase      ) if isinstance(UpperCAmelCase    ,							UpperCAmelCase      ) else bos_token
     __A    =  AddedToken(UpperCAmelCase    ,							lstrip=UpperCAmelCase    ,							rstrip=UpperCAmelCase      ) if isinstance(UpperCAmelCase    ,							UpperCAmelCase      ) else eos_token
     __A    =  AddedToken(UpperCAmelCase    ,							lstrip=UpperCAmelCase    ,							rstrip=UpperCAmelCase      ) if isinstance(UpperCAmelCase    ,							UpperCAmelCase      ) else sep_token
     __A    =  AddedToken(UpperCAmelCase    ,							lstrip=UpperCAmelCase    ,							rstrip=UpperCAmelCase      ) if isinstance(UpperCAmelCase    ,							UpperCAmelCase      ) else cls_token
     __A    =  AddedToken(UpperCAmelCase    ,							lstrip=UpperCAmelCase    ,							rstrip=UpperCAmelCase      ) if isinstance(UpperCAmelCase    ,							UpperCAmelCase      ) else unk_token
     __A    =  AddedToken(UpperCAmelCase    ,							lstrip=UpperCAmelCase    ,							rstrip=UpperCAmelCase      ) if isinstance(UpperCAmelCase    ,							UpperCAmelCase      ) else pad_token

     # Mask token behave like a normal word, i.e. include the space before it
     __A    =  AddedToken(UpperCAmelCase    ,							lstrip=UpperCAmelCase    ,							rstrip=UpperCAmelCase      ) if isinstance(UpperCAmelCase    ,							UpperCAmelCase      ) else mask_token

     super().__init__(
         errors=UpperCAmelCase    ,							bos_token=UpperCAmelCase    ,							eos_token=UpperCAmelCase    ,							unk_token=UpperCAmelCase    ,							sep_token=UpperCAmelCase    ,							cls_token=UpperCAmelCase    ,							pad_token=UpperCAmelCase    ,							mask_token=UpperCAmelCase    ,							add_prefix_space=UpperCAmelCase    ,							**UpperCAmelCase    ,							)
     with open(UpperCAmelCase    ,							encoding='''utf-8'''      ) as vocab_handle:
        __A    =  json.load(UpperCAmelCase      )
     __A    =  {v: k for k, v in self.encoder.items()}
     __A    =  errors  # how to handle errors in decoding
     __A    =  bytes_to_unicode()
     __A    =  {v: k for k, v in self.byte_encoder.items()}
     with open(UpperCAmelCase    ,							encoding='''utf-8'''      ) as merges_handle:
        __A    =  merges_handle.read().split('''\n'''      )[1:-1]
     __A    =  [tuple(merge.split()      ) for merge in bpe_merges]
     __A    =  dict(zip(UpperCAmelCase    ,							range(len(UpperCAmelCase      )      )      )      )
     __A    =  {}
     __A    =  add_prefix_space

     # Should have added re.IGNORECASE so BPE merges can happen for capitalized versions of contractions
     __A    =  re.compile(r'''\'s|\'t|\'re|\'ve|\'m|\'ll|\'d| ?\p{L}+| ?\p{N}+| ?[^\s\p{L}\p{N}]+|\s+(?!\S)|\s+'''      )

  @property
  def 						SCREAMING_SNAKE_CASE__			(   self      )->   Tuple:
     return len(self.encoder      )

  def 						SCREAMING_SNAKE_CASE__			(   self      )->   str:
     return dict(self.encoder    ,							**self.added_tokens_encoder      )

  def 						SCREAMING_SNAKE_CASE__			(   self    ,							UpperCAmelCase      )->   List[str]:
     if token in self.cache:
        return self.cache[token]
     __A    =  tuple(UpperCAmelCase      )
     __A    =  get_pairs(UpperCAmelCase      )

     if not pairs:
        return token

     while True:
        __A    =  min(UpperCAmelCase    ,							key=lambda UpperCAmelCase      : self.bpe_ranks.get(UpperCAmelCase    ,							float('''inf'''      )      )      )
        if bigram not in self.bpe_ranks:
           break
        __A   ,		__A    =  bigram
        __A    =  []
        __A    =  0
        while i < len(UpperCAmelCase      ):
           try:
              __A    =  word.index(UpperCAmelCase    ,							UpperCAmelCase      )
           except ValueError:
              new_word.extend(word[i:]      )
              break
           else:
              new_word.extend(word[i:j]      )
              __A    =  j

           if word[i] == first and i < len(UpperCAmelCase      ) - 1 and word[i + 1] == second:
              new_word.append(first + second      )
              i += 2
           else:
              new_word.append(word[i]      )
              i += 1
        __A    =  tuple(UpperCAmelCase      )
        __A    =  new_word
        if len(UpperCAmelCase      ) == 1:
           break
        else:
           __A    =  get_pairs(UpperCAmelCase      )
     __A    =  ''' '''.join(UpperCAmelCase      )
     __A    =  word
     return word

  def 						SCREAMING_SNAKE_CASE__			(   self    ,							UpperCAmelCase      )->   List[Any]:
     __A    =  []
     for token in re.findall(self.pat    ,							UpperCAmelCase      ):
        __A    =  ''''''.join(
            self.byte_encoder[b] for b in token.encode('''utf-8'''      )      )  # Maps all our bytes to unicode strings, avoiding control tokens of the BPE (spaces in our case)
        bpe_tokens.extend(bpe_token for bpe_token in self.bpe(UpperCAmelCase      ).split(''' '''      )      )
     return bpe_tokens

  def 						SCREAMING_SNAKE_CASE__			(   self    ,							UpperCAmelCase      )->   Any:
     return self.encoder.get(UpperCAmelCase    ,							self.encoder.get(self.unk_token      )      )

  def 						SCREAMING_SNAKE_CASE__			(   self    ,							UpperCAmelCase      )->   List[Any]:
     return self.decoder.get(UpperCAmelCase      )

  def 						SCREAMING_SNAKE_CASE__			(   self    ,							UpperCAmelCase      )->   Tuple:
     __A    =  ''''''.join(UpperCAmelCase      )
     __A    =  bytearray([self.byte_decoder[c] for c in text]      ).decode('''utf-8'''    ,							errors=self.errors      )
     return text

  def 						SCREAMING_SNAKE_CASE__			(   self    ,							UpperCAmelCase    ,							UpperCAmelCase = None      )->   Tuple[str]:
     if not os.path.isdir(UpperCAmelCase      ):
        logger.error(f"Vocabulary path ({save_directory}) should be a directory"      )
        return
     __A    =  os.path.join(
         UpperCAmelCase    ,							(filename_prefix + '''-''' if filename_prefix else '''''') + VOCAB_FILES_NAMES['''vocab_file''']      )
     __A    =  os.path.join(
         UpperCAmelCase    ,							(filename_prefix + '''-''' if filename_prefix else '''''') + VOCAB_FILES_NAMES['''merges_file''']      )

     with open(UpperCAmelCase    ,							'''w'''    ,							encoding='''utf-8'''      ) as f:
        f.write(json.dumps(self.encoder    ,							indent=2    ,							sort_keys=UpperCAmelCase    ,							ensure_ascii=UpperCAmelCase      ) + '''\n'''      )

     __A    =  0
     with open(UpperCAmelCase    ,							'''w'''    ,							encoding='''utf-8'''      ) as writer:
        writer.write('''#version: 0.2\n'''      )
        for bpe_tokens, token_index in sorted(self.bpe_ranks.items()    ,							key=lambda UpperCAmelCase      : kv[1]      ):
           if index != token_index:
              logger.warning(
                  f"Saving vocabulary to {merge_file}: BPE merge indices are not consecutive."
                  ''' Please check that the tokenizer is not corrupted!'''      )
              __A    =  token_index
           writer.write(''' '''.join(UpperCAmelCase      ) + '''\n'''      )
           index += 1

     return vocab_file, merge_file

  def 						SCREAMING_SNAKE_CASE__			(   self    ,							UpperCAmelCase    ,							UpperCAmelCase = None      )->   List[int]:
     if token_ids_a is None:
        return [self.cls_token_id] + token_ids_a + [self.sep_token_id]
     __A    =  [self.cls_token_id]
     __A    =  [self.sep_token_id]
     return cls + token_ids_a + sep + sep + token_ids_a + sep

  def 						SCREAMING_SNAKE_CASE__			(   self    ,							UpperCAmelCase    ,							UpperCAmelCase = None    ,							UpperCAmelCase = False      )->   List[int]:
     if already_has_special_tokens:
        return super().get_special_tokens_mask(
            token_ids_a=UpperCAmelCase    ,							token_ids_a=UpperCAmelCase    ,							already_has_special_tokens=UpperCAmelCase      )

     if token_ids_a is None:
        return [1] + ([0] * len(UpperCAmelCase      )) + [1]
     return [1] + ([0] * len(UpperCAmelCase      )) + [1, 1] + ([0] * len(UpperCAmelCase      )) + [1]

  def 						SCREAMING_SNAKE_CASE__			(   self    ,							UpperCAmelCase    ,							UpperCAmelCase = None      )->   List[int]:
     __A    =  [self.sep_token_id]
     __A    =  [self.cls_token_id]

     if token_ids_a is None:
        return len(cls + token_ids_a + sep      ) * [0]
     return len(cls + token_ids_a + sep + sep + token_ids_a + sep      ) * [0]





  def 						SCREAMING_SNAKE_CASE__			(   self    ,							UpperCAmelCase    ,							UpperCAmelCase=False    ,							**UpperCAmelCase      )->   Union[str, Any]:
     __A    =  kwargs.pop('''add_prefix_space'''    ,							self.add_prefix_space      )
     if (is_split_into_words or add_prefix_space) and (len(UpperCAmelCase      ) > 0 and not text[0].isspace()):
        __A    =  ''' ''' + text
     return (text, kwargs)

