





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


UpperCamelCase_	 ="""./tests/fixtures/elise_format0.mid"""




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






				def 							snake_case      (					self						:       int				)      -> str:




										'''simple docstring'''



										torch.manual_seed(0				)
										_UpperCamelCase     :   int      =       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''',					)

										_UpperCamelCase     :   Optional[Any]      =       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''',					)

										_UpperCamelCase     :   List[str]      =       TaFilmDecoder(
										    input_dims=1_2_8,					targets_length=2_5_6,					max_decoder_noise_time=20_000.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,					)

										_UpperCamelCase     :   Optional[Any]      =       DDPMScheduler()

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






				def 							snake_case      (					self						:       str,					lowerCAmelCase__						:       List[str],					lowerCAmelCase__						:       str=0				)      -> Optional[int]:




										'''simple docstring'''



										if str(lowerCAmelCase__				).startswith('''mps'''				):
																_UpperCamelCase     :   Union[str, Any]      =       torch.manual_seed(lowerCAmelCase__				)
										else:
																_UpperCamelCase     :   List[str]      =       torch.Generator(device=lowerCAmelCase__				).manual_seed(lowerCAmelCase__				)
										_UpperCamelCase     :   List[str]      =       {
										    '''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 							snake_case      (					self						:       List[Any]				)      -> Tuple:




										'''simple docstring'''



										_UpperCamelCase     :   Dict      =       '''cpu'''  # ensure determinism for the device-dependent torch.Generator
										_UpperCamelCase     :   Optional[int]      =       self.get_dummy_components()
										_UpperCamelCase     :   Tuple      =       SpectrogramDiffusionPipeline(**lowerCAmelCase__				)
										_UpperCamelCase     :   Dict      =       pipe.to(lowerCAmelCase__				)
										pipe.set_progress_bar_config(disable=lowerCAmelCase__				)

										_UpperCamelCase     :   Tuple      =       self.get_dummy_inputs(lowerCAmelCase__				)
										_UpperCamelCase     :   Dict      =       pipe(**lowerCAmelCase__				)
										_UpperCamelCase     :   Dict      =       output.audios

										_UpperCamelCase     :   Tuple      =       mel[0, -3:, -3:]

										assert mel_slice.shape == (3, 3)
										_UpperCamelCase     :   Tuple      =       np.array(
										    [-11.512_925, -4.788_215, -0.46_172_905, -2.051_715, -10.539_147, -10.970_963, -9.091_634, 4.0, 4.0]				)
										assert np.abs(mel_slice.flatten() - expected_slice				).max() < 1e-2






				@skip_mps
				def 							snake_case      (					self						:       Union[str, Any]				)      -> List[Any]:




										'''simple docstring'''



										return super().test_save_load_local()






				@skip_mps
				def 							snake_case      (					self						:       str				)      -> Union[str, Any]:




										'''simple docstring'''



										return super().test_dict_tuple_outputs_equivalent()






				@skip_mps
				def 							snake_case      (					self						:       List[Any]				)      -> int:




										'''simple docstring'''



										return super().test_save_load_optional_components()






				@skip_mps
				def 							snake_case      (					self						:       int				)      -> Optional[int]:




										'''simple docstring'''



										return super().test_attention_slicing_forward_pass()






				def 							snake_case      (					self						:       Any				)      -> Dict:




										'''simple docstring'''



										pass






				def 							snake_case      (					self						:       List[str]				)      -> List[Any]:




										'''simple docstring'''



										pass






				@skip_mps
				def 							snake_case      (					self						:       int				)      -> List[Any]:




										'''simple docstring'''



										return super().test_progress_bar()




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






				def 							snake_case      (					self						:       Optional[int]				)      -> List[str]:




										'''simple docstring'''



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






				def 							snake_case      (					self						:       int				)      -> Optional[Any]:




										'''simple docstring'''



										_UpperCamelCase     :   List[Any]      =       torch_device

										_UpperCamelCase     :   Union[str, Any]      =       SpectrogramDiffusionPipeline.from_pretrained('''google/music-spectrogram-diffusion'''				)
										_UpperCamelCase     :   Dict      =       pipe.melgan
										_UpperCamelCase     :   Tuple      =       None

										_UpperCamelCase     :   Tuple      =       pipe.to(lowerCAmelCase__				)
										pipe.set_progress_bar_config(disable=lowerCAmelCase__				)

										def callback(lowerCAmelCase__						:       int,					lowerCAmelCase__						:       Optional[Any]				):
																# decode mel to audio
																_UpperCamelCase     :   Optional[Any]      =       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

										_UpperCamelCase     :   Dict      =       MidiProcessor()
										_UpperCamelCase     :   Tuple      =       processor(lowerCAmelCase__				)

										_UpperCamelCase     :   Optional[int]      =       input_tokens[:3]
										_UpperCamelCase     :   List[Any]      =       torch.manual_seed(0				)
										pipe(lowerCAmelCase__,					num_inference_steps=5,					generator=lowerCAmelCase__,					callback=lowerCAmelCase__,					output_type='''mel'''				)






				def 							snake_case      (					self						:       Optional[Any]				)      -> Optional[Any]:




										'''simple docstring'''



										_UpperCamelCase     :   Dict      =       torch_device

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

										_UpperCamelCase     :   Tuple      =       processor(lowerCAmelCase__				)
										# just run two denoising loops
										_UpperCamelCase     :   Tuple      =       input_tokens[:2]

										_UpperCamelCase     :   List[str]      =       torch.manual_seed(0				)
										_UpperCamelCase     :   Any      =       pipe(lowerCAmelCase__,					num_inference_steps=2,					generator=lowerCAmelCase__				)

										_UpperCamelCase     :   Dict      =       output.audios[0]

										assert abs(np.abs(lowerCAmelCase__				).sum() - 3_612.841				) < 1e-1






				def 							snake_case      (					self						:       Tuple				)      -> List[str]:




										'''simple docstring'''



										_UpperCamelCase     :   int      =       torch_device

										_UpperCamelCase     :   Tuple      =       SpectrogramDiffusionPipeline.from_pretrained('''google/music-spectrogram-diffusion'''				)
										_UpperCamelCase     :   Union[str, Any]      =       pipe.to(lowerCAmelCase__				)
										pipe.set_progress_bar_config(disable=lowerCAmelCase__				)

										_UpperCamelCase     :   List[str]      =       MidiProcessor()

										_UpperCamelCase     :   Tuple      =       processor(lowerCAmelCase__				)

										# just run 4 denoising loops
										_UpperCamelCase     :   Tuple      =       input_tokens[:4]

										_UpperCamelCase     :   List[Any]      =       torch.manual_seed(0				)
										_UpperCamelCase     :   Dict      =       pipe(lowerCAmelCase__,					num_inference_steps=1_0_0,					generator=lowerCAmelCase__				)

										_UpperCamelCase     :   Optional[int]      =       output.audios[0]
										assert abs(np.abs(lowerCAmelCase__				).sum() - 9_389.1_111				) < 5e-2






import os
import sys
from pathlib import Path

import pytest

from datasets import Dataset, IterableDataset
from datasets.distributed import split_dataset_by_node

from .utils import execute_subprocess_async, get_torch_dist_unique_port, require_torch




def      UpperCAmelCase__							(     ):





  '''simple docstring'''







  a__    =Dataset.from_dict({'''i''': range(17				)}				)
  a__    =len(_A				)
  a__    =3
  a__    =[
      split_dataset_by_node(_A      , rank=_A      , world_size=_A				) for rank in range(_A				)
  ]
  assert sum(len(_A				) for ds in datasets_per_rank				) == full_size
  assert len({tuple(x.values()				) for ds in datasets_per_rank for x in ds}				) == full_size




def      UpperCAmelCase__							(     ):





  '''simple docstring'''







  def gen():
    return ({"i": i} for i in range(17				))

  a__    =3
  a__    =IterableDataset.from_generator(_A				)
  a__    =len(list(_A				)				)
  a__    =[
      split_dataset_by_node(_A      , rank=_A      , world_size=_A				) for rank in range(_A				)
  ]
  assert sum(len(list(_A				)				) for ds in datasets_per_rank				) == full_size
  assert len({tuple(x.values()				) for ds in datasets_per_rank for x in ds}				) == full_size




@pytest.mark.parametrize('''shards_per_node'''      , [1, 2, 3]				)
def      UpperCAmelCase__							(     _A :					Dict				):





  '''simple docstring'''







  def gen(_A :					List[str]				):
    for shard in shards:
      yield from ({"i": i, "shard": shard} for i in range(17				))

  a__    =3
  a__    =shards_per_node * world_size
  a__    ={'''shards''': [F"""shard_{shard_idx}.txt""" for shard_idx in range(_A				)]}
  a__    =IterableDataset.from_generator(_A      , gen_kwargs=_A				)
  a__    =len(list(_A				)				)
  assert full_ds.n_shards == world_size * shards_per_node
  a__    =[
      split_dataset_by_node(_A      , rank=_A      , world_size=_A				) for rank in range(_A				)
  ]
  assert [ds.n_shards for ds in datasets_per_rank] == [shards_per_node] * world_size
  assert sum(len(list(_A				)				) for ds in datasets_per_rank				) == full_size
  assert len({tuple(x.values()				) for ds in datasets_per_rank for x in ds}				) == full_size




def      UpperCAmelCase__							(     ):





  '''simple docstring'''







  def gen():
    return ({"i": i} for i in range(17				))

  a__    =2
  a__    =IterableDataset.from_generator(_A				)
  a__    =len(list(_A				)				)

  a__    =split_dataset_by_node(_A      , rank=0      , world_size=_A				).shuffle(seed=42				)
  assert len(list(_A				)				) == 1 + full_size // world_size
  with pytest.raises(_A				):
    split_dataset_by_node(_A      , rank=0      , world_size=_A				).shuffle()

  a__    =split_dataset_by_node(full_ds.shuffle(seed=42				)      , rank=0      , world_size=_A				)
  assert len(list(_A				)				) == 1 + full_size // world_size
  with pytest.raises(_A				):
    split_dataset_by_node(full_ds.shuffle()      , rank=0      , world_size=_A				)




@pytest.mark.parametrize('''streaming'''      , [False, True]				)
@require_torch
@pytest.mark.skipif(os.name == '''nt'''      , reason='''execute_subprocess_async doesn\'t support windows'''				)
@pytest.mark.integration
def      UpperCAmelCase__							(     _A :					Dict				):





  '''simple docstring'''







  a__    =2
  a__    =get_torch_dist_unique_port()
  a__    =Path(_A				).resolve().parent / '''distributed_scripts''' / '''launch_torch_distributed.py'''
  a__    =F"""
        -m torch.distributed.launch
        --nproc_per_node={nproc_per_node}
        --master_port={master_port}
        {test_script}
    """.split()
  a__    =F"""
        --streaming={streaming}
    """.split()
  a__    =[sys.executable] + distributed_args + args
  execute_subprocess_async(_A      , env=os.environ.copy()				)




@pytest.mark.parametrize(
    '''nproc_per_node, num_workers'''      , [
        (2, 2),  # each node has 2 shards and each worker has 1 shards
        (3, 2),  # each node uses all the shards but skips examples, and each worker has 2 shards
    ]      , )
@require_torch
@pytest.mark.skipif(os.name == '''nt'''      , reason='''execute_subprocess_async doesn\'t support windows'''				)
@pytest.mark.integration
def      UpperCAmelCase__							(     _A :					List[str]      , _A :					str				):





  '''simple docstring'''







  a__    =True
  a__    =get_torch_dist_unique_port()
  a__    =Path(_A				).resolve().parent / '''distributed_scripts''' / '''launch_torch_distributed.py'''
  a__    =F"""
        -m torch.distributed.launch
        --nproc_per_node={nproc_per_node}
        --master_port={master_port}
        {test_script}
    """.split()
  a__    =F"""
        --streaming={streaming}
        --num_workers={num_workers}
    """.split()
  a__    =[sys.executable] + distributed_args + args
  execute_subprocess_async(_A      , env=os.environ.copy()				)
