



import gc
import unittest

import numpy as np
import torch

from diffusers import DanceDiffusionPipeline, IPNDMScheduler, UNetaDModel
from diffusers.utils import slow, torch_device
from diffusers.utils.testing_utils import enable_full_determinism, require_torch_gpu, skip_mps

from ..pipeline_params import UNCONDITIONAL_AUDIO_GENERATION_BATCH_PARAMS, UNCONDITIONAL_AUDIO_GENERATION_PARAMS
from ..test_pipelines_common import PipelineTesterMixin


enable_full_determinism()


class       _lowercase   (a_      ,   unittest.TestCase      ):
				'''simple docstring'''







				lowercase__   						=						DanceDiffusionPipeline
				lowercase__   						=						UNCONDITIONAL_AUDIO_GENERATION_PARAMS
				lowercase__   						=						PipelineTesterMixin.required_optional_params - {
				    """callback""",
				    """latents""",
				    """callback_steps""",
				    """output_type""",
				    """num_images_per_prompt""",
				}
				lowercase__   						=						UNCONDITIONAL_AUDIO_GENERATION_BATCH_PARAMS
				lowercase__   						=						False
				lowercase__   						=						False

				def       _lowerCamelCase					( self    ):

									'''simple docstring'''


									torch.manual_seed(0    )
									UpperCamelCase_		  =   UNetaDModel(
									    block_out_channels=(32, 32, 64)     , extra_in_channels=16     , sample_size=512     , sample_rate=1_6000     , in_channels=2     , out_channels=2     , flip_sin_to_cos=snake_case__     , use_timestep_embedding=snake_case__     , time_embedding_type="fourier"     , mid_block_type="UNetMidBlock1D"     , down_block_types=("DownBlock1DNoSkip", "DownBlock1D", "AttnDownBlock1D")     , up_block_types=("AttnUpBlock1D", "UpBlock1D", "UpBlock1DNoSkip")     , )
									UpperCamelCase_		  =   IPNDMScheduler()

									UpperCamelCase_		  =   {
									    "unet": unet,
									    "scheduler": scheduler,
									}
									return components

				def       _lowerCamelCase					( self     , snake_case__     , snake_case__=0    ):

									'''simple docstring'''


									if str(snake_case__    ).startswith("mps"    ):
														UpperCamelCase_		  =   torch.manual_seed(snake_case__    )
									else:
														UpperCamelCase_		  =   torch.Generator(device=snake_case__    ).manual_seed(snake_case__    )
									UpperCamelCase_		  =   {
									    "batch_size": 1,
									    "generator": generator,
									    "num_inference_steps": 4,
									}
									return inputs

				def       _lowerCamelCase					( self    ):

									'''simple docstring'''


									UpperCamelCase_		  =   "cpu"  # ensure determinism for the device-dependent torch.Generator
									UpperCamelCase_		  =   self.get_dummy_components()
									UpperCamelCase_		  =   DanceDiffusionPipeline(**snake_case__    )
									UpperCamelCase_		  =   pipe.to(snake_case__    )
									pipe.set_progress_bar_config(disable=snake_case__    )

									UpperCamelCase_		  =   self.get_dummy_inputs(snake_case__    )
									UpperCamelCase_		  =   pipe(**snake_case__    )
									UpperCamelCase_		  =   output.audios

									UpperCamelCase_		  =   audio[0, -3:, -3:]

									assert audio.shape == (1, 2, components["unet"].sample_size)
									UpperCamelCase_		  =   np.array([-0.7_265, 1.0_000, -0.8_388, 0.1_175, 0.9_498, -1.0_000]    )
									assert np.abs(audio_slice.flatten() - expected_slice    ).max() < 1e-2

				@skip_mps
				def       _lowerCamelCase					( self    ):

									'''simple docstring'''


									return super().test_save_load_local()

				@skip_mps
				def       _lowerCamelCase					( self    ):

									'''simple docstring'''


									return super().test_dict_tuple_outputs_equivalent(expected_max_difference=3e-3    )

				@skip_mps
				def       _lowerCamelCase					( self    ):

									'''simple docstring'''


									return super().test_save_load_optional_components()

				@skip_mps
				def       _lowerCamelCase					( self    ):

									'''simple docstring'''


									return super().test_attention_slicing_forward_pass()

				def       _lowerCamelCase					( self    ):

									'''simple docstring'''


									super().test_inference_batch_single_identical(expected_max_diff=3e-3    )




@slow
@require_torch_gpu
class       _lowercase   (unittest.TestCase      ):
				'''simple docstring'''







				def       _lowerCamelCase					( self    ):

									'''simple docstring'''


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

				def       _lowerCamelCase					( self    ):

									'''simple docstring'''


									UpperCamelCase_		  =   torch_device

									UpperCamelCase_		  =   DanceDiffusionPipeline.from_pretrained("harmonai/maestro-150k"    )
									UpperCamelCase_		  =   pipe.to(snake_case__    )
									pipe.set_progress_bar_config(disable=snake_case__    )

									UpperCamelCase_		  =   torch.manual_seed(0    )
									UpperCamelCase_		  =   pipe(generator=snake_case__     , num_inference_steps=100     , audio_length_in_s=4.096    )
									UpperCamelCase_		  =   output.audios

									UpperCamelCase_		  =   audio[0, -3:, -3:]

									assert audio.shape == (1, 2, pipe.unet.sample_size)
									UpperCamelCase_		  =   np.array([-0.0_192, -0.0_231, -0.0_318, -0.0_059, 0.0_002, -0.0_020]    )

									assert np.abs(audio_slice.flatten() - expected_slice    ).max() < 1e-2

				def       _lowerCamelCase					( self    ):

									'''simple docstring'''


									UpperCamelCase_		  =   torch_device

									UpperCamelCase_		  =   DanceDiffusionPipeline.from_pretrained("harmonai/maestro-150k"     , torch_dtype=torch.floataa    )
									UpperCamelCase_		  =   pipe.to(snake_case__    )
									pipe.set_progress_bar_config(disable=snake_case__    )

									UpperCamelCase_		  =   torch.manual_seed(0    )
									UpperCamelCase_		  =   pipe(generator=snake_case__     , num_inference_steps=100     , audio_length_in_s=4.096    )
									UpperCamelCase_		  =   output.audios

									UpperCamelCase_		  =   audio[0, -3:, -3:]

									assert audio.shape == (1, 2, pipe.unet.sample_size)
									UpperCamelCase_		  =   np.array([-0.0_367, -0.0_488, -0.0_771, -0.0_525, -0.0_444, -0.0_341]    )

									assert np.abs(audio_slice.flatten() - expected_slice    ).max() < 1e-2





import torch



def    _lowerCAmelCase    ():
					if torch.cuda.is_available():
										UpperCamelCase_		  =   torch.cuda.device_count()
					else:
										UpperCamelCase_		  =   0
					print(f"""Successfully ran on {num_gpus} GPUs""")


if __name__ == "__main__":
	main()
