






"""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 					lowerCamelCase__					(       snake_case    ,     unittest.TestCase ):
    SCREAMING_SNAKE_CASE		=			SpectrogramDiffusionPipeline
    SCREAMING_SNAKE_CASE		=			PipelineTesterMixin.required_optional_params - {
        '''callback''',
        '''latents''',
        '''callback_steps''',
        '''output_type''',
        '''num_images_per_prompt''',
    }
    SCREAMING_SNAKE_CASE		=			False
    SCREAMING_SNAKE_CASE		=			False
    SCREAMING_SNAKE_CASE		=			TOKENS_TO_AUDIO_GENERATION_PARAMS
    SCREAMING_SNAKE_CASE		=			TOKENS_TO_AUDIO_GENERATION_BATCH_PARAMS






    def 				_UpperCamelCase							( self					):
          torch.manual_seed(0					)
          UpperCAmelCase							=       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"""	,)

          UpperCAmelCase							=       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"""	,)

          UpperCAmelCase							=       TaFilmDecoder(
              input_dims=128	,targets_length=256	,max_decoder_noise_time=20000.0	,d_model=768	,num_layers=1	,num_heads=1	,d_kv=4	,d_ff=2_048	,dropout_rate=0.1	,)

          UpperCAmelCase							=       DDPMScheduler()

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






    def 				_UpperCamelCase							( self	,A	,A=0					):
          if str(A					).startswith("""mps"""					):
                UpperCAmelCase							=       torch.manual_seed(A					)
          else:
                UpperCAmelCase							=       torch.Generator(device=A					).manual_seed(A					)
          UpperCAmelCase							=       {
              """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 				_UpperCamelCase							( self					):
          UpperCAmelCase							=       """cpu"""  # ensure determinism for the device-dependent torch.Generator
          UpperCAmelCase							=       self.get_dummy_components()
          UpperCAmelCase							=       SpectrogramDiffusionPipeline(**A					)
          UpperCAmelCase							=       pipe.to(A					)
          pipe.set_progress_bar_config(disable=A					)

          UpperCAmelCase							=       self.get_dummy_inputs(A					)
          UpperCAmelCase							=       pipe(**A					)
          UpperCAmelCase							=       output.audios

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

          assert mel_slice.shape == (3, 3)
          UpperCAmelCase							=       np.array(
              [-11.512925, -4.788215, -0.46172905, -2.051715, -10.539147, -10.970963, -9.091634, 4.0, 4.0]					)
          assert np.abs(mel_slice.flatten() - expected_slice					).max() < 1e-2






    @skip_mps
    def 				_UpperCamelCase							( self					):
          return super().test_save_load_local()






    @skip_mps
    def 				_UpperCamelCase							( self					):
          return super().test_dict_tuple_outputs_equivalent()






    @skip_mps
    def 				_UpperCamelCase							( self					):
          return super().test_save_load_optional_components()






    @skip_mps
    def 				_UpperCamelCase							( self					):
          return super().test_attention_slicing_forward_pass()






    def 				_UpperCamelCase							( self					):
          pass






    def 				_UpperCamelCase							( self					):
          pass






    @skip_mps
    def 				_UpperCamelCase							( self					):
          return super().test_progress_bar()







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






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






    def 				_UpperCamelCase							( self					):
          # TODO - test that pipeline can decode tokens in a callback
          # so that music can be played live
          UpperCAmelCase							=       torch_device

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

          UpperCAmelCase							=       pipe.to(A					)
          pipe.set_progress_bar_config(disable=A					)

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

          UpperCAmelCase							=       MidiProcessor()
          UpperCAmelCase							=       processor(A					)

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






    def 				_UpperCamelCase							( self					):
          UpperCAmelCase							=       torch_device

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

          UpperCAmelCase							=       processor(A					)
          # just run two denoising loops
          UpperCAmelCase							=       input_tokens[:2]

          UpperCAmelCase							=       torch.manual_seed(0					)
          UpperCAmelCase							=       pipe(A	,num_inference_steps=2	,generator=A					)

          UpperCAmelCase							=       output.audios[0]

          assert abs(np.abs(A					).sum() - 3612.841					) < 1e-1






    def 				_UpperCamelCase							( self					):
          UpperCAmelCase							=       torch_device

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

          UpperCAmelCase							=       MidiProcessor()

          UpperCAmelCase							=       processor(A					)

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

          UpperCAmelCase							=       torch.manual_seed(0					)
          UpperCAmelCase							=       pipe(A	,num_inference_steps=100	,generator=A					)

          UpperCAmelCase							=       output.audios[0]
          assert abs(np.abs(A					).sum() - 9389.1111					) < 5e-2











"""simple docstring"""







import sys

def _a (		_snake_case      ,       _snake_case							):







      """simple docstring"""






      if not numbers:
            return 0
      if target == 0 and target in numbers:
            return 0
      if not isinstance(_snake_case      ,       (list, tuple)							) or not all(
          isinstance(_snake_case      ,       _snake_case							) for number in numbers							):
            raise ValueError("""numbers must be an iterable of integers"""							)

      UpperCAmelCase							=       UpperCAmelCase							=       UpperCAmelCase							=       0
      UpperCAmelCase							=       sys.maxsize

      while right < len(_snake_case							):
            curr_sum += numbers[right]
            while curr_sum >= target and left <= right:
                  UpperCAmelCase							=       min(_snake_case      ,       right - left + 1							)
                  curr_sum -= numbers[left]
                  left += 1
            right += 1

      return 0 if min_len == sys.maxsize else min_len



