
'''simple docstring'''





import importlib
import importlib.metadata
import os
import warnings
from distutils.util import strtobool
from functools import lru_cache

import torch
from packaging import version
from packaging.version import parse

from .environment import parse_flag_from_env
from .versions import compare_versions, is_torch_version


try:
     import torch_xla.core.xla_model as xm  # noqa: F401

     A							     = True
except ImportError:
     A							     = False


# Cache this result has it's a C FFI call which can be pretty time-consuming
A							     = torch.distributed.is_available()
def 	snake_case_							(				_a  :				int      ):
      # Check we're not importing a "pkg_name" directory somewhere but the actual library by trying to grab the version
      snake_case_    							=				importlib.util.find_spec(_a      ) is not None
      if package_exists:
            try:
                  snake_case_    							=				importlib.metadata.metadata(_a      )
                  return True
            except importlib.metadata.PackageNotFoundError:
                  return False
def 	snake_case_							(				):
      return _torch_distributed_available
def 	snake_case_							(				):
      try:
            pass
      except ImportError:
            print(
                "Intel(R) oneCCL Bindings for PyTorch* is required to run DDP on Intel(R) GPUs, but it is not"
                " detected. If you see \"ValueError: Invalid backend: 'ccl'\" error, please install Intel(R) oneCCL"
                " Bindings for PyTorch*."      )
      return (
          importlib.util.find_spec("torch_ccl"      ) is not None
          or importlib.util.find_spec("oneccl_bindings_for_pytorch"      ) is not None
      )
def 	snake_case_							(				):
      return importlib.metadata.version("oneccl_bind_pt"      )
def 	snake_case_							(				):
      return _is_package_available("transformer_engine"      )
@lru_cache
def 	snake_case_							(				_a  :				str=True      ):
      # Due to bugs on the amp series GPUs, we disable torch-xla on them
      if torch.cuda.is_available():
            return False
      if _tpu_available and check_device:
            try:
                  # Will raise a RuntimeError if no XLA configuration is found
                  snake_case_    							=				xm.xla_device()
                  return True
            except RuntimeError:
                  return False
      return _tpu_available
def 	snake_case_							(				):
      return _is_package_available("deepspeed"      )
def 	snake_case_							(				_a  :				List[str]=False      ):
      if is_tpu_available():
            return not ignore_tpu
      if torch.cuda.is_available():
            return torch.cuda.is_bfaa_supported()
      if is_npu_available():
            return False
      return True
def 	snake_case_							(				):
      snake_case_    							=				_is_package_available("bitsandbytes"      )
      if package_exists:
            snake_case_    							=				version.parse(importlib.metadata.version("bitsandbytes"      )      )
            return compare_versions(_a			,       ">="			,       "0.39.0"      )
      return False
def 	snake_case_							(				):
      snake_case_    							=				_is_package_available("bitsandbytes"      )
      if package_exists:
            snake_case_    							=				version.parse(importlib.metadata.version("bitsandbytes"      )      )
            return compare_versions(_a			,       ">="			,       "0.37.2"      )
      return False
def 	snake_case_							(				):
      return _is_package_available("bitsandbytes"      )
def 	snake_case_							(				):
      if strtobool(os.environ.get("ACCELERATE_USE_MEGATRON_LM"			,       "False"      )      ) == 1:
            snake_case_    							=				importlib.util.find_spec("megatron"      ) is not None
            if package_exists:
                  try:
                        snake_case_    							=				parse(importlib.metadata.version("megatron-lm"      )      )
                        return compare_versions(_a			,       ">="			,       "2.2.0"      )
                  except Exception as e:
                        warnings.warn(F'''Parse Megatron version failed. Exception:{e}'''      )
                        return False
def 	snake_case_							(				):
      return _is_package_available("safetensors"      )
def 	snake_case_							(				):
      return _is_package_available("transformers"      )
def 	snake_case_							(				):
      return _is_package_available("datasets"      )
def 	snake_case_							(				):
      return _is_package_available("aim"      )
def 	snake_case_							(				):
      return _is_package_available("tensorboard"      ) or _is_package_available("tensorboardX"      )
def 	snake_case_							(				):
      return _is_package_available("wandb"      )
def 	snake_case_							(				):
      return _is_package_available("comet_ml"      )
def 	snake_case_							(				):
      return _is_package_available("boto3"      )
def 	snake_case_							(				):
      if _is_package_available("rich"      ):
            if "ACCELERATE_DISABLE_RICH" in os.environ:
                  warnings.warn(
                      "`ACCELERATE_DISABLE_RICH` is deprecated and will be removed in v0.22.0 and deactivated by default. Please use `ACCELERATE_ENABLE_RICH` if you wish to use `rich`."      )
                  return not parse_flag_from_env("ACCELERATE_DISABLE_RICH"			,       _a      )
            return parse_flag_from_env("ACCELERATE_ENABLE_RICH"			,       _a      )
      return False
def 	snake_case_							(				):
      return _is_package_available("sagemaker"      )
def 	snake_case_							(				):
      return _is_package_available("tqdm"      )
def 	snake_case_							(				):
      return _is_package_available("mlflow"      )
def 	snake_case_							(				):
      return is_torch_version(">="			,       "1.12"      ) and torch.backends.mps.is_available() and torch.backends.mps.is_built()
def 	snake_case_							(				):
      def get_major_and_minor_from_version(_a  :				Union[str, Any]      ):
            return str(version.parse(_a      ).major      ) + "." + str(version.parse(_a      ).minor      )

      snake_case_    							=				importlib.metadata.version("torch"      )
      if importlib.util.find_spec("intel_extension_for_pytorch"      ) is None:
            return False
      snake_case_    							=				"N/A"
      try:
            snake_case_    							=				importlib.metadata.version("intel_extension_for_pytorch"      )
      except importlib.metadata.PackageNotFoundError:
            return False
      snake_case_    							=				get_major_and_minor_from_version(_torch_version      )
      snake_case_    							=				get_major_and_minor_from_version(_ipex_version      )
      if torch_major_and_minor != ipex_major_and_minor:
            warnings.warn(
                F'''Intel Extension for PyTorch {ipex_major_and_minor} needs to work with PyTorch {ipex_major_and_minor}.*,'''
                F''' but PyTorch {_torch_version} is found. Please switch to the matching version and run again.'''      )
            return False
      return True
@lru_cache
def 	snake_case_							(				_a  :				Union[str, Any]=False      ):
      if importlib.util.find_spec("torch"      ) is None or importlib.util.find_spec("torch_npu"      ) is None:
            return False

      import torch
      import torch_npu  # noqa: F401

      if check_device:
            try:
                  # Will raise a RuntimeError if no NPU is found
                  snake_case_    							=				torch.npu.device_count()
                  return torch.npu.is_available()
            except RuntimeError:
                  return False
      return hasattr(_a			,       "npu"      ) and torch.npu.is_available()




@lru_cache
def 	snake_case_							(				_a  :				Optional[Any]=False      ):
      if not parse_flag_from_env("ACCELERATE_USE_XPU"			,       default=_a      ):
            return False
      if is_ipex_available():
            import torch

            if is_torch_version("<="			,       "1.12"      ):
                  return False
      else:
            return False

      import intel_extension_for_pytorch  # noqa: F401

      if check_device:
            try:
                  # Will raise a RuntimeError if no XPU  is found
                  snake_case_    							=				torch.xpu.device_count()
                  return torch.xpu.is_available()
            except RuntimeError:
                  return False
      return hasattr(_a			,       "xpu"      ) and torch.xpu.is_available()







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