

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     snake_case_			(				_a			, unittest.TestCase							):


							"""simple docstring"""





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




							def        A__  ( self							):
												torch.manual_seed(0							)
												__lowerCAmelCase					=							UNetaDModel(
												    block_out_channels=(3_2, 3_2, 6_4)					,						extra_in_channels=1_6					,						sample_size=5_1_2					,						sample_rate=1_6_0_0_0					,						in_channels=2					,						out_channels=2					,						flip_sin_to_cos=_A					,						use_timestep_embedding=_A					,						time_embedding_type='fourier'					,						mid_block_type='UNetMidBlock1D'					,						down_block_types=('DownBlock1DNoSkip', 'DownBlock1D', 'AttnDownBlock1D')					,						up_block_types=('AttnUpBlock1D', 'UpBlock1D', 'UpBlock1DNoSkip')					,						)
												__lowerCAmelCase					=							IPNDMScheduler()

												__lowerCAmelCase					=							{
												    'unet': unet,
												    'scheduler': scheduler,
												}
												return components




							def        A__  ( self					,						_A					,						_A=0							):
												if str(_A							).startswith('mps'							):
																	__lowerCAmelCase					=							torch.manual_seed(_A							)
												else:
																	__lowerCAmelCase					=							torch.Generator(device=_A							).manual_seed(_A							)
												__lowerCAmelCase					=							{
												    'batch_size': 1,
												    'generator': generator,
												    'num_inference_steps': 4,
												}
												return inputs




							def        A__  ( self							):
												__lowerCAmelCase					=							'cpu'  # ensure determinism for the device-dependent torch.Generator
												__lowerCAmelCase					=							self.get_dummy_components()
												__lowerCAmelCase					=							DanceDiffusionPipeline(**_A							)
												__lowerCAmelCase					=							pipe.to(_A							)
												pipe.set_progress_bar_config(disable=_A							)

												__lowerCAmelCase					=							self.get_dummy_inputs(_A							)
												__lowerCAmelCase					=							pipe(**_A							)
												__lowerCAmelCase					=							output.audios

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

												assert audio.shape == (1, 2, components["unet"].sample_size)
												__lowerCAmelCase					=							np.array([-0.7265, 1.0000, -0.8388, 0.1175, 0.9498, -1.0000]							)
												assert np.abs(audio_slice.flatten() - expected_slice							).max() < 1e-2




							@skip_mps
							def        A__  ( self							):
												return super().test_save_load_local()




							@skip_mps
							def        A__  ( self							):
												return super().test_dict_tuple_outputs_equivalent(expected_max_difference=3e-3							)




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




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




							def        A__  ( self							):
												super().test_inference_batch_single_identical(expected_max_diff=3e-3							)





@slow
@require_torch_gpu
class     snake_case_			(				unittest.TestCase							):


							"""simple docstring"""





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




							def        A__  ( self							):
												__lowerCAmelCase					=							torch_device

												__lowerCAmelCase					=							DanceDiffusionPipeline.from_pretrained('harmonai/maestro-150k'							)
												__lowerCAmelCase					=							pipe.to(_A							)
												pipe.set_progress_bar_config(disable=_A							)

												__lowerCAmelCase					=							torch.manual_seed(0							)
												__lowerCAmelCase					=							pipe(generator=_A					,						num_inference_steps=1_0_0					,						audio_length_in_s=4.096							)
												__lowerCAmelCase					=							output.audios

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

												assert audio.shape == (1, 2, pipe.unet.sample_size)
												__lowerCAmelCase					=							np.array([-0.0192, -0.0231, -0.0318, -0.0059, 0.0002, -0.0020]							)

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




							def        A__  ( self							):
												__lowerCAmelCase					=							torch_device

												__lowerCAmelCase					=							DanceDiffusionPipeline.from_pretrained('harmonai/maestro-150k'					,						torch_dtype=torch.floataa							)
												__lowerCAmelCase					=							pipe.to(_A							)
												pipe.set_progress_bar_config(disable=_A							)

												__lowerCAmelCase					=							torch.manual_seed(0							)
												__lowerCAmelCase					=							pipe(generator=_A					,						num_inference_steps=1_0_0					,						audio_length_in_s=4.096							)
												__lowerCAmelCase					=							output.audios

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

												assert audio.shape == (1, 2, pipe.unet.sample_size)
												__lowerCAmelCase					=							np.array([-0.0367, -0.0488, -0.0771, -0.0525, -0.0444, -0.0341]							)

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









from typing import TYPE_CHECKING

from ...utils import (
    OptionalDependencyNotAvailable,
    _LazyModule,
    is_flax_available,
    is_tf_available,
    is_torch_available,
)


lowerCamelCase      				=     {'''configuration_unispeech''': ['''UNISPEECH_PRETRAINED_CONFIG_ARCHIVE_MAP''', '''UniSpeechConfig''']}

try:
		if not is_torch_available():
				raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
		pass
else:
		lowerCamelCase      				=     [
		    '''UNISPEECH_PRETRAINED_MODEL_ARCHIVE_LIST''',
		    '''UniSpeechForCTC''',
		    '''UniSpeechForPreTraining''',
		    '''UniSpeechForSequenceClassification''',
		    '''UniSpeechModel''',
		    '''UniSpeechPreTrainedModel''',
		]

if TYPE_CHECKING:
		from .configuration_unispeech import UNISPEECH_PRETRAINED_CONFIG_ARCHIVE_MAP, UniSpeechConfig

		try:
				if not is_torch_available():
						raise OptionalDependencyNotAvailable()
		except OptionalDependencyNotAvailable:
				pass
		else:
				from .modeling_unispeech import (
				    UNISPEECH_PRETRAINED_MODEL_ARCHIVE_LIST,
				    UniSpeechForCTC,
				    UniSpeechForPreTraining,
				    UniSpeechForSequenceClassification,
				    UniSpeechModel,
				    UniSpeechPreTrainedModel,
				)

else:
		import sys

		lowerCamelCase      				=     _LazyModule(__name__, globals()['''__file__'''], _import_structure, module_spec=__spec__)






