
'''simple docstring'''
import warnings
from typing import List, Optional, Union

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

from .configuration_utils import ConfigMixin, register_to_config
from .utils import CONFIG_NAME, PIL_INTERPOLATION, deprecate






class 				UpperCAmelCase_			(		snake_case__   ):
 UpperCAmelCase_						      =  CONFIG_NAME





 @register_to_config
 def __init__(      self	,    lowercase_ = True	,    lowercase_ = 8	,    lowercase_ = "lanczos"	,    lowercase_ = True	,    lowercase_ = False	,    ):
     super().__init__()





 @staticmethod
 def     snake_case__						(      lowercase_):
     if images.ndim == 3:
         snake_case_	:      Any            =  images[None, ...]
     snake_case_	:      Union[str, Any]            =  (images * 2_55).round().astype("uint8")
     if images.shape[-1] == 1:
         # special case for grayscale (single channel) images
         snake_case_	:      Union[str, Any]            =  [Image.fromarray(image.squeeze()	,    mode="L") for image in images]
     else:
         snake_case_	:      Dict            =  [Image.fromarray(lowercase_) for image in images]

     return pil_images





 @staticmethod
 def     snake_case__						(      lowercase_):
     if not isinstance(lowercase_	,    lowercase_):
         snake_case_	:      str            =  [images]
     snake_case_	:      str            =  [np.array(lowercase_).astype(np.floataa) / 255.0 for image in images]
     snake_case_	:      Dict            =  np.stack(lowercase_	,    axis=0)

     return images





 @staticmethod
 def     snake_case__						(      lowercase_):
     if images.ndim == 3:
         snake_case_	:      Any            =  images[..., None]

     snake_case_	:      int            =  torch.from_numpy(images.transpose(0	,    3	,    1	,    2))
     return images





 @staticmethod
 def     snake_case__						(      lowercase_):
     snake_case_	:      str            =  images.cpu().permute(0	,    2	,    3	,    1).float().numpy()
     return images





 @staticmethod
 def     snake_case__						(      lowercase_):
     return 2.0 * images - 1.0





 @staticmethod
 def     snake_case__						(      lowercase_):
     return (images / 2 + 0.5).clamp(0	,    1)





 @staticmethod
 def     snake_case__						(      lowercase_):
     snake_case_	:      Optional[int]            =  image.convert("RGB")
     return image





 def     snake_case__						(      self	,    lowercase_	,    lowercase_ = None	,    lowercase_ = None	,    ):
     if height is None:
         snake_case_	:      str            =  image.height
     if width is None:
         snake_case_	:      Dict            =  image.width

     snake_case_	:      List[Any]            =  (
         x - x % self.config.vae_scale_factor for x in (width, height)
     )  # resize to integer multiple of vae_scale_factor
     snake_case_	:      int            =  image.resize((width, height)	,    resample=PIL_INTERPOLATION[self.config.resample])
     return image





 def     snake_case__						(      self	,    lowercase_	,    lowercase_ = None	,    lowercase_ = None	,    ):
     snake_case_	:      Optional[int]            =  (PIL.Image.Image, np.ndarray, torch.Tensor)
     if isinstance(lowercase_	,    lowercase_):
         snake_case_	:      Optional[Any]            =  [image]
     elif not (isinstance(lowercase_	,    lowercase_) and all(isinstance(lowercase_	,    lowercase_) for i in image)):
         raise ValueError(
             F'Input is in incorrect format: {[type(lowercase_) for i in image]}. Currently, we only support {", ".join(lowercase_)}')

     if isinstance(image[0]	,    PIL.Image.Image):
         if self.config.do_convert_rgb:
             snake_case_	:      Union[str, Any]            =  [self.convert_to_rgb(lowercase_) for i in image]
         if self.config.do_resize:
             snake_case_	:      int            =  [self.resize(lowercase_	,    lowercase_	,    lowercase_) for i in image]
         snake_case_	:      Dict            =  self.pil_to_numpy(lowercase_)  # to np
         snake_case_	:      List[Any]            =  self.numpy_to_pt(lowercase_)  # to pt

     elif isinstance(image[0]	,    np.ndarray):
         snake_case_	:      Optional[int]            =  np.concatenate(lowercase_	,    axis=0) if image[0].ndim == 4 else np.stack(lowercase_	,    axis=0)
         snake_case_	:      str            =  self.numpy_to_pt(lowercase_)
         snake_case_	:      List[str]            =  image.shape
         if self.config.do_resize and (
             height % self.config.vae_scale_factor != 0 or width % self.config.vae_scale_factor != 0
         ):
             raise ValueError(
                 F'Currently we only support resizing for PIL image - please resize your numpy array to be divisible by {self.config.vae_scale_factor}'
                 F'currently the sizes are {height} and {width}. You can also pass a PIL image instead to use resize option in VAEImageProcessor')

     elif isinstance(image[0]	,    torch.Tensor):
         snake_case_	:      List[str]            =  torch.cat(lowercase_	,    axis=0) if image[0].ndim == 4 else torch.stack(lowercase_	,    axis=0)
         snake_case_	:      List[str]            =  image.shape

         # don't need any preprocess if the image is latents
         if channel == 4:
             return image

         if self.config.do_resize and (
             height % self.config.vae_scale_factor != 0 or width % self.config.vae_scale_factor != 0
         ):
             raise ValueError(
                 F'Currently we only support resizing for PIL image - please resize your pytorch tensor to be divisible by {self.config.vae_scale_factor}'
                 F'currently the sizes are {height} and {width}. You can also pass a PIL image instead to use resize option in VAEImageProcessor')

        # expected range [0,1], normalize to [-1,1]
     snake_case_	:      Optional[Any]            =  self.config.do_normalize
     if image.min() < 0:
         warnings.warn(
             "Passing `image` as torch tensor with value range in [-1,1] is deprecated. The expected value range for image tensor is [0,1] "
             F'when passing as pytorch tensor or numpy Array. You passed `image` with value range [{image.min()},{image.max()}]'	,    lowercase_	,    )
         snake_case_	:      Tuple            =  False

     if do_normalize:
         snake_case_	:      int            =  self.normalize(lowercase_)

     return image





 def     snake_case__						(      self	,    lowercase_	,    lowercase_ = "pil"	,    lowercase_ = None	,    ):
     if not isinstance(lowercase_	,    torch.Tensor):
         raise ValueError(
             F'Input for postprocessing is in incorrect format: {type(lowercase_)}. We only support pytorch tensor')
     if output_type not in ["latent", "pt", "np", "pil"]:
         snake_case_	:      Tuple            =  (
             F'the output_type {output_type} is outdated and has been set to `np`. Please make sure to set it to one of these instead: '
             "`pil`, `np`, `pt`, `latent`"
         )
         deprecate("Unsupported output_type"	,    "1.0.0"	,    lowercase_	,    standard_warn=lowercase_)
         snake_case_	:      Dict            =  "np"

     if output_type == "latent":
         return image

     if do_denormalize is None:
         snake_case_	:      Optional[Any]            =  [self.config.do_normalize] * image.shape[0]

     snake_case_	:      Tuple            =  torch.stack(
         [self.denormalize(image[i]) if do_denormalize[i] else image[i] for i in range(image.shape[0])])

     if output_type == "pt":
         return image

     snake_case_	:      Optional[int]            =  self.pt_to_numpy(lowercase_)

     if output_type == "np":
         return image

     if output_type == "pil":
         return self.numpy_to_pil(lowercase_)






class 				UpperCAmelCase_			(		snake_case__   ):
 UpperCAmelCase_						      =  CONFIG_NAME





 @register_to_config
 def __init__(      self	,    lowercase_ = True	,    lowercase_ = 8	,    lowercase_ = "lanczos"	,    lowercase_ = True	,    ):
     super().__init__()





 @staticmethod
 def     snake_case__						(      lowercase_):
     if images.ndim == 3:
         snake_case_	:      Optional[int]            =  images[None, ...]
     snake_case_	:      str            =  (images * 2_55).round().astype("uint8")
     if images.shape[-1] == 1:
         # special case for grayscale (single channel) images
         snake_case_	:      Any            =  [Image.fromarray(image.squeeze()	,    mode="L") for image in images]
     else:
         snake_case_	:      str            =  [Image.fromarray(image[:, :, :3]) for image in images]

     return pil_images





 @staticmethod
 def     snake_case__						(      lowercase_):
     return image[:, :, 1] * 2**8 + image[:, :, 2]





 def     snake_case__						(      self	,    lowercase_):
     if images.ndim == 3:
         snake_case_	:      List[Any]            =  images[None, ...]
     snake_case_	:      Any            =  images[:, :, :, 3:]
     if images.shape[-1] == 6:
         snake_case_	:      Optional[int]            =  (images_depth * 2_55).round().astype("uint8")
         snake_case_	:      Optional[Any]            =  [
             Image.fromarray(self.rgblike_to_depthmap(lowercase_)	,    mode="I;16") for image_depth in images_depth
         ]
     elif images.shape[-1] == 4:
         snake_case_	:      Optional[Any]            =  (images_depth * 6_55_35.0).astype(np.uintaa)
         snake_case_	:      List[str]            =  [Image.fromarray(lowercase_	,    mode="I;16") for image_depth in images_depth]
     else:
         raise Exception("Not supported")

     return pil_images





 def     snake_case__						(      self	,    lowercase_	,    lowercase_ = "pil"	,    lowercase_ = None	,    ):
     if not isinstance(lowercase_	,    torch.Tensor):
         raise ValueError(
             F'Input for postprocessing is in incorrect format: {type(lowercase_)}. We only support pytorch tensor')
     if output_type not in ["latent", "pt", "np", "pil"]:
         snake_case_	:      Optional[int]            =  (
             F'the output_type {output_type} is outdated and has been set to `np`. Please make sure to set it to one of these instead: '
             "`pil`, `np`, `pt`, `latent`"
         )
         deprecate("Unsupported output_type"	,    "1.0.0"	,    lowercase_	,    standard_warn=lowercase_)
         snake_case_	:      Any            =  "np"

     if do_denormalize is None:
         snake_case_	:      Dict            =  [self.config.do_normalize] * image.shape[0]

     snake_case_	:      Optional[int]            =  torch.stack(
         [self.denormalize(image[i]) if do_denormalize[i] else image[i] for i in range(image.shape[0])])

     snake_case_	:      Tuple            =  self.pt_to_numpy(lowercase_)

     if output_type == "np":
         if image.shape[-1] == 6:
             snake_case_	:      List[str]            =  np.stack([self.rgblike_to_depthmap(im[:, :, 3:]) for im in image]	,    axis=0)
         else:
             snake_case_	:      Tuple            =  image[:, :, :, 3:]
         return image[:, :, :, :3], image_depth

     if output_type == "pil":
         return self.numpy_to_pil(lowercase_), self.numpy_to_depth(lowercase_)
     else:
         raise Exception(F'This type {output_type} is not supported')






'''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 				UpperCAmelCase_			(		snake_case__      ,  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):
     torch.manual_seed(0)
     snake_case_	:      List[str]            =  SpectrogramNotesEncoder(
         max_length=20_48	,    vocab_size=15_36	,    d_model=7_68	,    dropout_rate=0.1	,    num_layers=1	,    num_heads=1	,    d_kv=4	,    d_ff=20_48	,    feed_forward_proj="gated-gelu"	,    )

     snake_case_	:      int            =  SpectrogramContEncoder(
         input_dims=1_28	,    targets_context_length=2_56	,    d_model=7_68	,    dropout_rate=0.1	,    num_layers=1	,    num_heads=1	,    d_kv=4	,    d_ff=20_48	,    feed_forward_proj="gated-gelu"	,    )

     snake_case_	:      Tuple            =  TaFilmDecoder(
         input_dims=1_28	,    targets_length=2_56	,    max_decoder_noise_time=20_000.0	,    d_model=7_68	,    num_layers=1	,    num_heads=1	,    d_kv=4	,    d_ff=20_48	,    dropout_rate=0.1	,    )

     snake_case_	:      Union[str, Any]            =  DDPMScheduler()

     snake_case_	:      List[Any]            =  {
         "notes_encoder": notes_encoder.eval(),
         "continuous_encoder": continuous_encoder.eval(),
         "decoder": decoder.eval(),
         "scheduler": scheduler,
         "melgan": None,
     }
     return components





 def     snake_case__						(      self	,    lowercase_	,    lowercase_=0):
     if str(lowercase_).startswith("mps"):
         snake_case_	:      Dict            =  torch.manual_seed(lowercase_)
     else:
         snake_case_	:      Dict            =  torch.Generator(device=lowercase_).manual_seed(lowercase_)
     snake_case_	:      int            =  {
         "input_tokens": [
             [11_34, 90, 11_35, 11_33, 10_80, 1_12, 11_32, 10_80, 11_33, 10_79, 1_33, 11_32, 10_79, 11_33, 1] + [0] * 20_33
         ],
         "generator": generator,
         "num_inference_steps": 4,
         "output_type": "mel",
     }
     return inputs





 def     snake_case__						(      self):
     snake_case_	:      List[Any]            =  "cpu"  # ensure determinism for the device-dependent torch.Generator
     snake_case_	:      List[Any]            =  self.get_dummy_components()
     snake_case_	:      Tuple            =  SpectrogramDiffusionPipeline(**lowercase_)
     snake_case_	:      Tuple            =  pipe.to(lowercase_)
     pipe.set_progress_bar_config(disable=lowercase_)

     snake_case_	:      List[str]            =  self.get_dummy_inputs(lowercase_)
     snake_case_	:      Any            =  pipe(**lowercase_)
     snake_case_	:      str            =  output.audios

     snake_case_	:      str            =  mel[0, -3:, -3:]

     assert mel_slice.shape == (3, 3)
     snake_case_	:      Any            =  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):
     return super().test_save_load_local()





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





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





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





 def     snake_case__						(      self):
     pass





 def     snake_case__						(      self):
     pass





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






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





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





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

     snake_case_	:      str            =  SpectrogramDiffusionPipeline.from_pretrained("google/music-spectrogram-diffusion")
     snake_case_	:      Tuple            =  pipe.melgan
     snake_case_	:      List[str]            =  None

     snake_case_	:      int            =  pipe.to(lowercase_)
     pipe.set_progress_bar_config(disable=lowercase_)

     def callback(lowercase_	,    lowercase_):
         # decode mel to audio
         snake_case_	:      Dict            =  melgan(input_features=mel_output.astype(np.floataa))[0]
         assert len(audio[0]) == 8_19_20 * (step + 1)
         # simulate that audio is played
         return audio

     snake_case_	:      Tuple            =  MidiProcessor()
     snake_case_	:      int            =  processor(lowercase_)

     snake_case_	:      int            =  input_tokens[:3]
     snake_case_	:      Union[str, Any]            =  torch.manual_seed(0)
     pipe(lowercase_	,    num_inference_steps=5	,    generator=lowercase_	,    callback=lowercase_	,    output_type="mel")





 def     snake_case__						(      self):
     snake_case_	:      Any            =  torch_device

     snake_case_	:      Optional[Any]            =  SpectrogramDiffusionPipeline.from_pretrained("google/music-spectrogram-diffusion")
     snake_case_	:      Optional[int]            =  pipe.to(lowercase_)
     pipe.set_progress_bar_config(disable=lowercase_)
     snake_case_	:      Union[str, Any]            =  MidiProcessor()

     snake_case_	:      List[str]            =  processor(lowercase_)
     # just run two denoising loops
     snake_case_	:      int            =  input_tokens[:2]

     snake_case_	:      Optional[Any]            =  torch.manual_seed(0)
     snake_case_	:      Tuple            =  pipe(lowercase_	,    num_inference_steps=2	,    generator=lowercase_)

     snake_case_	:      Optional[Any]            =  output.audios[0]

     assert abs(np.abs(lowercase_).sum() - 3_612.841) < 1E-1





 def     snake_case__						(      self):
     snake_case_	:      Union[str, Any]            =  torch_device

     snake_case_	:      Tuple            =  SpectrogramDiffusionPipeline.from_pretrained("google/music-spectrogram-diffusion")
     snake_case_	:      Optional[Any]            =  pipe.to(lowercase_)
     pipe.set_progress_bar_config(disable=lowercase_)

     snake_case_	:      Tuple            =  MidiProcessor()

     snake_case_	:      int            =  processor(lowercase_)

     # just run 4 denoising loops
     snake_case_	:      Any            =  input_tokens[:4]

     snake_case_	:      Optional[Any]            =  torch.manual_seed(0)
     snake_case_	:      Tuple            =  pipe(lowercase_	,    num_inference_steps=1_00	,    generator=lowercase_)

     snake_case_	:      Tuple            =  output.audios[0]
     assert abs(np.abs(lowercase_).sum() - 9_389.1_111) < 5E-2




