


'''simple docstring'''

import sys
from typing import Tuple

import numpy as np
import torch
from PIL import Image
from torch import nn

from transformers.image_utils import PILImageResampling
from utils import img_tensorize







class 			UpperCAmelCase			:

 def __init__(			self			:						Union[str, Any],       a_			:						Optional[Any],       a_			:						Optional[Any]=sys.maxsize       ):





    """simple docstring"""



    UpperCamelCase__			    =    "bilinear"
    UpperCamelCase__			    =    max_size
    UpperCamelCase__			    =    short_edge_length

 def __call__(			self			:						int,       a_			:						int       ):





    """simple docstring"""



    UpperCamelCase__			    =    []
    for img in imgs:
       UpperCamelCase__    ,		UpperCamelCase__			    =    img.shape[:2]
       # later: provide list and randomly choose index for resize
       UpperCamelCase__			    =    np.random.randint(self.short_edge_length[0],       self.short_edge_length[1] + 1       )
       if size == 0:
          return img
       UpperCamelCase__			    =    size * 1.0 / min(a_,       a_       )
       if h < w:
          UpperCamelCase__    ,		UpperCamelCase__			    =    size, scale * w
       else:
          UpperCamelCase__    ,		UpperCamelCase__			    =    scale * h, size
       if max(a_,       a_       ) > self.max_size:
          UpperCamelCase__			    =    self.max_size * 1.0 / max(a_,       a_       )
          UpperCamelCase__			    =    newh * scale
          UpperCamelCase__			    =    neww * scale
       UpperCamelCase__			    =    int(neww + 0.5       )
       UpperCamelCase__			    =    int(newh + 0.5       )

       if img.dtype == np.uinta:
          UpperCamelCase__			    =    Image.fromarray(a_       )
          UpperCamelCase__			    =    pil_image.resize((neww, newh),       PILImageResampling.BILINEAR       )
          UpperCamelCase__			    =    np.asarray(a_       )
       else:
          UpperCamelCase__			    =    img.permute(2,       0,       1       ).unsqueeze(0       )  # 3, 0, 1)  # hw(c) -> nchw
          UpperCamelCase__			    =    nn.functional.interpolate(
              a_,       (newh, neww),       mode=self.interp_method,       align_corners=a_       ).squeeze(0       )
       img_augs.append(a_       )

    return img_augs







class 			UpperCAmelCase			:

 def __init__(			self			:						int,       a_			:						Dict       ):





    """simple docstring"""



    UpperCamelCase__			    =    ResizeShortestEdge([cfg.INPUT.MIN_SIZE_TEST, cfg.INPUT.MIN_SIZE_TEST],       cfg.INPUT.MAX_SIZE_TEST       )
    UpperCamelCase__			    =    cfg.INPUT.FORMAT
    UpperCamelCase__			    =    cfg.SIZE_DIVISIBILITY
    UpperCamelCase__			    =    cfg.PAD_VALUE
    UpperCamelCase__			    =    cfg.INPUT.MAX_SIZE_TEST
    UpperCamelCase__			    =    cfg.MODEL.DEVICE
    UpperCamelCase__			    =    torch.tensor(cfg.MODEL.PIXEL_STD       ).to(self.device       ).view(len(cfg.MODEL.PIXEL_STD       ),       1,       1       )
    UpperCamelCase__			    =    torch.tensor(cfg.MODEL.PIXEL_MEAN       ).to(self.device       ).view(len(cfg.MODEL.PIXEL_STD       ),       1,       1       )
    UpperCamelCase__			    =    lambda a_       : (x - self.pixel_mean) / self.pixel_std

 def 		lowercase_    (			self			:						Dict,       a_			:						Tuple       ):





    """simple docstring"""



    UpperCamelCase__			    =    tuple(max(a_       ) for s in zip(*[img.shape for img in images]       )       )
    UpperCamelCase__			    =    [im.shape[-2:] for im in images]
    UpperCamelCase__			    =    [
        nn.functional.pad(
            a_,       [0, max_size[-1] - size[1], 0, max_size[-2] - size[0]],       value=self.pad_value,       )
        for size, im in zip(a_,       a_       )
    ]

    return torch.stack(a_       ), torch.tensor(a_       )

 def __call__(			self			:						List[Any],       a_			:						Optional[int],       a_			:						List[Any]=False       ):





    """simple docstring"""



    with torch.no_grad():
       if not isinstance(a_,       a_       ):
          UpperCamelCase__			    =    [images]
       if single_image:
          assert len(a_       ) == 1
       for i in range(len(a_       )       ):
          if isinstance(images[i],       torch.Tensor       ):
             images.insert(a_,       images.pop(a_       ).to(self.device       ).float()       )
          elif not isinstance(images[i],       torch.Tensor       ):
             images.insert(
                 a_,       torch.as_tensor(img_tensorize(images.pop(a_       ),       input_format=self.input_format       )       )
                 .to(self.device       )
                 .float(),       )
            # resize smallest edge
       UpperCamelCase__			    =    torch.tensor([im.shape[:2] for im in images]       )
       UpperCamelCase__			    =    self.aug(a_       )
       # transpose images and convert to torch tensors
       # images = [torch.as_tensor(i.astype("float32")).permute(2, 0, 1).to(self.device) for i in images]
       # now normalize before pad to avoid useless arithmetic
       UpperCamelCase__			    =    [self.normalizer(a_       ) for x in images]
       # now pad them to do the following operations
       UpperCamelCase__    ,		UpperCamelCase__			    =    self.pad(a_       )
       # Normalize

       if self.size_divisibility > 0:
          raise NotImplementedError()
       # pad
       UpperCamelCase__			    =    torch.true_divide(a_,       a_       )
       if single_image:
          return images[0], sizes[0], scales_yx[0]
       else:
          return images, sizes, scales_yx







def 							SCREAMING_SNAKE_CASE__(		_UpperCamelCase   :					Dict     ,		_UpperCamelCase   :					int )		->       str:





   '''simple docstring'''







   boxes[:, 0::2] *= scale_yx[:, 1]
   boxes[:, 1::2] *= scale_yx[:, 0]
   return boxes






def 							SCREAMING_SNAKE_CASE__(		_UpperCamelCase   :					int     ,		_UpperCamelCase   :					Tuple[int, int] )		->       int:





   '''simple docstring'''







   assert torch.isfinite(_UpperCamelCase ).all(), "Box tensor contains infinite or NaN!"
   UpperCamelCase__    ,		UpperCamelCase__			    =    box_size
   tensor[:, 0].clamp_(min=0     ,		max=_UpperCamelCase )
   tensor[:, 1].clamp_(min=0     ,		max=_UpperCamelCase )
   tensor[:, 2].clamp_(min=0     ,		max=_UpperCamelCase )
   tensor[:, 3].clamp_(min=0     ,		max=_UpperCamelCase )





'''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