




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  :		int		     =   '''./tests/fixtures/elise_format0.mid'''







@unittest.skip("""The note-seq package currently throws an error on import"""					)
class 				UpperCAmelCase(					snake_case_   ,     unittest.TestCase					):



	"""simple docstring"""






	a     : int	  =					SpectrogramDiffusionPipeline
	a     : int	  =					PipelineTesterMixin.required_optional_params - {
	    """callback""",
	    """latents""",
	    """callback_steps""",
	    """output_type""",
	    """num_images_per_prompt""",
	}
	a     : Optional[Any]	  =					False
	a     : List[Any]	  =					False
	a     : Dict	  =					TOKENS_TO_AUDIO_GENERATION_PARAMS
	a     : Union[str, Any]	  =					TOKENS_TO_AUDIO_GENERATION_BATCH_PARAMS

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





								"""simple docstring"""


								torch.manual_seed(0    )
								lowercase__  :	Optional[int]        =							SpectrogramNotesEncoder(
								    max_length=2048	,				vocab_size=1536	,				d_model=768	,				dropout_rate=0.1	,				num_layers=1	,				num_heads=1	,				d_kv=4	,				d_ff=2048	,				feed_forward_proj="gated-gelu"	,				)

								lowercase__  :	int        =							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=2048	,				feed_forward_proj="gated-gelu"	,				)

								lowercase__  :	Any        =							TaFilmDecoder(
								    input_dims=128	,				targets_length=256	,				max_decoder_noise_time=2_00_00.0	,				d_model=768	,				num_layers=1	,				num_heads=1	,				d_kv=4	,				d_ff=2048	,				dropout_rate=0.1	,				)

								lowercase__  :	Optional[Any]        =							DDPMScheduler()

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

	def    __a     (					self	,				lowerCamelCase	,				lowerCamelCase=0    ) ->      List[Any]:





								"""simple docstring"""


								if str(lowerCamelCase    ).startswith("mps"    ):
															lowercase__  :	Optional[int]        =							torch.manual_seed(lowerCamelCase    )
								else:
															lowercase__  :	Optional[Any]        =							torch.Generator(device=lowerCamelCase    ).manual_seed(lowerCamelCase    )
								lowercase__  :	str        =							{
								    "input_tokens": [
								        [1134, 90, 1135, 1133, 1080, 112, 1132, 1080, 1133, 1079, 133, 1132, 1079, 1133, 1] + [0] * 2033
								    ],
								    "generator": generator,
								    "num_inference_steps": 4,
								    "output_type": "mel",
								}
								return inputs

	def    __a     (					self    ) ->      Tuple:





								"""simple docstring"""


								lowercase__  :	Dict        =							"cpu"  # ensure determinism for the device-dependent torch.Generator
								lowercase__  :	List[str]        =							self.get_dummy_components()
								lowercase__  :	Union[str, Any]        =							SpectrogramDiffusionPipeline(**lowerCamelCase    )
								lowercase__  :	Optional[int]        =							pipe.to(lowerCamelCase    )
								pipe.set_progress_bar_config(disable=lowerCamelCase    )

								lowercase__  :	str        =							self.get_dummy_inputs(lowerCamelCase    )
								lowercase__  :	str        =							pipe(**lowerCamelCase    )
								lowercase__  :	List[str]        =							output.audios

								lowercase__  :	Any        =							mel[0, -3:, -3:]

								assert mel_slice.shape == (3, 3)
								lowercase__  :	Union[str, Any]        =							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    __a     (					self    ) ->      List[str]:





								"""simple docstring"""


								return super().test_save_load_local()

	@skip_mps
	def    __a     (					self    ) ->      str:





								"""simple docstring"""


								return super().test_dict_tuple_outputs_equivalent()

	@skip_mps
	def    __a     (					self    ) ->      Any:





								"""simple docstring"""


								return super().test_save_load_optional_components()

	@skip_mps
	def    __a     (					self    ) ->      Any:





								"""simple docstring"""


								return super().test_attention_slicing_forward_pass()

	def    __a     (					self    ) ->      Any:





								"""simple docstring"""


								pass

	def    __a     (					self    ) ->      Tuple:





								"""simple docstring"""


								pass







	@skip_mps
	def    __a     (					self    ) ->      str:





								"""simple docstring"""


								return super().test_progress_bar()







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



	"""simple docstring"""






	def    __a     (					self    ) ->      Optional[int]:





								"""simple docstring"""


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

	def    __a     (					self    ) ->      Tuple:





								"""simple docstring"""


								lowercase__  :	Union[str, Any]        =							torch_device

								lowercase__  :	Union[str, Any]        =							SpectrogramDiffusionPipeline.from_pretrained("google/music-spectrogram-diffusion"    )
								lowercase__  :	List[str]        =							pipe.melgan
								lowercase__  :	Optional[Any]        =							None

								lowercase__  :	List[str]        =							pipe.to(lowerCamelCase    )
								pipe.set_progress_bar_config(disable=lowerCamelCase    )

								def callback(lowerCamelCase	,				lowerCamelCase    ):
															# decode mel to audio
															lowercase__  :	List[str]        =							melgan(input_features=mel_output.astype(np.floataa    )    )[0]
															assert len(audio[0]    ) == 81920 * (step + 1)
															# simulate that audio is played
															return audio

								lowercase__  :	str        =							MidiProcessor()
								lowercase__  :	str        =							processor(lowerCamelCase    )

								lowercase__  :	Tuple        =							input_tokens[:3]
								lowercase__  :	Dict        =							torch.manual_seed(0    )
								pipe(lowerCamelCase	,				num_inference_steps=5	,				generator=lowerCamelCase	,				callback=lowerCamelCase	,				output_type="mel"    )

	def    __a     (					self    ) ->      str:





								"""simple docstring"""


								lowercase__  :	Any        =							torch_device

								lowercase__  :	Optional[Any]        =							SpectrogramDiffusionPipeline.from_pretrained("google/music-spectrogram-diffusion"    )
								lowercase__  :	List[str]        =							pipe.to(lowerCamelCase    )
								pipe.set_progress_bar_config(disable=lowerCamelCase    )
								lowercase__  :	Any        =							MidiProcessor()

								lowercase__  :	int        =							processor(lowerCamelCase    )
								# just run two denoising loops
								lowercase__  :	Optional[Any]        =							input_tokens[:2]

								lowercase__  :	Any        =							torch.manual_seed(0    )
								lowercase__  :	Optional[Any]        =							pipe(lowerCamelCase	,				num_inference_steps=2	,				generator=lowerCamelCase    )

								lowercase__  :	Tuple        =							output.audios[0]

								assert abs(np.abs(lowerCamelCase    ).sum() - 36_12.8_41    ) < 1E-1







	def    __a     (					self    ) ->      Any:





								"""simple docstring"""


								lowercase__  :	Dict        =							torch_device

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

								lowercase__  :	Union[str, Any]        =							MidiProcessor()

								lowercase__  :	List[str]        =							processor(lowerCamelCase    )

								# just run 4 denoising loops
								lowercase__  :	List[str]        =							input_tokens[:4]

								lowercase__  :	Tuple        =							torch.manual_seed(0    )
								lowercase__  :	str        =							pipe(lowerCamelCase	,				num_inference_steps=100	,				generator=lowerCamelCase    )

								lowercase__  :	Union[str, Any]        =							output.audios[0]
								assert abs(np.abs(lowerCamelCase    ).sum() - 93_89.11_11    ) < 5E-2





import hashlib
import importlib.util
import json
import os
import pathlib
from types import ModuleType

import pytest
import requests

__a  :		str		     =   pathlib.Path.cwd().joinpath('''project_euler''')
__a  :		Tuple		     =   pathlib.Path.cwd().joinpath(
    '''scripts''', '''project_euler_answers.json'''
)

with open(PROJECT_EULER_ANSWERS_PATH) as file_handle:
						__a  :		dict[str, str]		     =   json.load(file_handle)
def  snake_case_					(		SCREAMING_SNAKE_CASE_    )      ->			ModuleType:
							lowercase__  :	List[str]        =							importlib.util.spec_from_file_location(file_path.name    ,str(SCREAMING_SNAKE_CASE_    )    )
							lowercase__  :	int        =							importlib.util.module_from_spec(SCREAMING_SNAKE_CASE_    )  # type: ignore
							spec.loader.exec_module(SCREAMING_SNAKE_CASE_    )  # type: ignore
							return module
def  snake_case_					(		)      ->			list[pathlib.Path]:
							lowercase__  :	List[str]        =							[]
							for problem_dir_path in PROJECT_EULER_DIR_PATH.iterdir():
														if problem_dir_path.is_file() or problem_dir_path.name.startswith("_"    ):
																					continue
														for file_path in problem_dir_path.iterdir():
																					if file_path.suffix != ".py" or file_path.name.startswith(("_", "test")    ):
																												continue
																					solution_file_paths.append(SCREAMING_SNAKE_CASE_    )
							return solution_file_paths
def  snake_case_					(		)      ->			str:
							with open(os.environ["GITHUB_EVENT_PATH"]    ) as file:
														lowercase__  :	str        =							json.load(SCREAMING_SNAKE_CASE_    )
							return event["pull_request"]["url"] + "/files"
def  snake_case_					(		)      ->			list[pathlib.Path]:
							lowercase__  :	Optional[Any]        =							[]
							lowercase__  :	Dict        =							{
							    "Accept": "application/vnd.github.v3+json",
							    "Authorization": "token " + os.environ["GITHUB_TOKEN"],
							}
							lowercase__  :	Union[str, Any]        =							requests.get(get_files_url()    ,headers=SCREAMING_SNAKE_CASE_    ).json()
							for file in files:
														lowercase__  :	Optional[Any]        =							pathlib.Path.cwd().joinpath(file["filename"]    )
														if (
														    filepath.suffix != ".py"
														    or filepath.name.startswith(("_", "test")    )
														    or not filepath.name.startswith("sol"    )
														):
																					continue
														solution_file_paths.append(SCREAMING_SNAKE_CASE_    )
							return solution_file_paths
def  snake_case_					(		)      ->			list[pathlib.Path]:
							if os.environ.get("CI"    ) and os.environ.get("GITHUB_EVENT_NAME"    ) == "pull_request":
														# Return only if there are any, otherwise default to all solutions
														if filepaths := added_solution_file_path():
																					return filepaths
							return all_solution_file_paths()




@pytest.mark.parametrize(
    "solution_path"    ,collect_solution_file_paths()    ,ids=lambda SCREAMING_SNAKE_CASE_    : F"""{path.parent.name}/{path.name}"""    ,)
def  snake_case_					(		SCREAMING_SNAKE_CASE_    )      ->			None:
							lowercase__  :	str        =							solution_path.parent.name[8:].zfill(3    )
							lowercase__  :	str        =							PROBLEM_ANSWERS[problem_number]
							lowercase__  :	Optional[Any]        =							convert_path_to_module(SCREAMING_SNAKE_CASE_    )
							lowercase__  :	str        =							str(solution_module.solution()    )  # type: ignore
							lowercase__  :	List[str]        =							hashlib.shaaaa(answer.encode()    ).hexdigest()
							assert (
							    answer == expected
							), F"""Expected solution to {problem_number} to have hash {expected}, got {answer}"""