


'''simple docstring'''




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  lowerCAmelCase				(				a      ,				unittest.TestCase							):
	_lowerCamelCase   :					Union[str, Any]			      = DanceDiffusionPipeline
	_lowerCamelCase   :					Tuple			      = UNCONDITIONAL_AUDIO_GENERATION_PARAMS
	_lowerCamelCase   :					Any			      = PipelineTesterMixin.required_optional_params - {
	    """callback""",
	    """latents""",
	    """callback_steps""",
	    """output_type""",
	    """num_images_per_prompt""",
	}
	_lowerCamelCase   :					Optional[Any]			      = UNCONDITIONAL_AUDIO_GENERATION_BATCH_PARAMS
	_lowerCamelCase   :					str			      = False
	_lowerCamelCase   :					Tuple			      = False



	def 				lowercase     (					self    ):
		torch.manual_seed(0    )
		lowerCAmelCase					:    List[str]      							= 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=UpperCamelCase_ ,					use_timestep_embedding=UpperCamelCase_ ,					time_embedding_type='fourier' ,					mid_block_type='UNetMidBlock1D' ,					down_block_types=('DownBlock1DNoSkip', 'DownBlock1D', 'AttnDownBlock1D') ,					up_block_types=('AttnUpBlock1D', 'UpBlock1D', 'UpBlock1DNoSkip') ,					)
		lowerCAmelCase					:    List[str]      							= IPNDMScheduler()

		lowerCAmelCase					:    Tuple      							= {
		    'unet': unet,
		    'scheduler': scheduler,
		}
		return components



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





		if str(UpperCamelCase_    ).startswith('mps'    ):
			lowerCAmelCase					:    str      							= torch.manual_seed(UpperCamelCase_    )
		else:
			lowerCAmelCase					:    str      							= torch.Generator(device=UpperCamelCase_    ).manual_seed(UpperCamelCase_    )
		lowerCAmelCase					:    Optional[int]      							= {
		    'batch_size': 1,
		    'generator': generator,
		    'num_inference_steps': 4,
		}
		return inputs



	def 				lowercase     (					self    ):
		lowerCAmelCase					:    Union[str, Any]      							= 'cpu'  # ensure determinism for the device-dependent torch.Generator
		lowerCAmelCase					:    int      							= self.get_dummy_components()
		lowerCAmelCase					:    Optional[int]      							= DanceDiffusionPipeline(**UpperCamelCase_    )
		lowerCAmelCase					:    int      							= pipe.to(UpperCamelCase_    )
		pipe.set_progress_bar_config(disable=UpperCamelCase_    )

		lowerCAmelCase					:    str      							= self.get_dummy_inputs(UpperCamelCase_    )
		lowerCAmelCase					:    Optional[Any]      							= pipe(**UpperCamelCase_    )
		lowerCAmelCase					:    Any      							= output.audios

		lowerCAmelCase					:    Dict      							= audio[0, -3:, -3:]

		assert audio.shape == (1, 2, components["unet"].sample_size)
		lowerCAmelCase					:    Union[str, Any]      							= np.array([-0.7_2_6_5, 1.0_0_0_0, -0.8_3_8_8, 0.1_1_7_5, 0.9_4_9_8, -1.0_0_0_0]    )
		assert np.abs(audio_slice.flatten() - expected_slice    ).max() < 1e-2



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



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



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



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



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



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



	def 				lowercase     (					self    ):
		super().tearDown()
		gc.collect()
		torch.cuda.empty_cache()



	def 				lowercase     (					self    ):
		lowerCAmelCase					:    int      							= torch_device

		lowerCAmelCase					:    Tuple      							= DanceDiffusionPipeline.from_pretrained('harmonai/maestro-150k'    )
		lowerCAmelCase					:    Dict      							= pipe.to(UpperCamelCase_    )
		pipe.set_progress_bar_config(disable=UpperCamelCase_    )

		lowerCAmelCase					:    int      							= torch.manual_seed(0    )
		lowerCAmelCase					:    Dict      							= pipe(generator=UpperCamelCase_ ,					num_inference_steps=100 ,					audio_length_in_s=4.0_9_6    )
		lowerCAmelCase					:    Union[str, Any]      							= output.audios

		lowerCAmelCase					:    Union[str, Any]      							= audio[0, -3:, -3:]

		assert audio.shape == (1, 2, pipe.unet.sample_size)
		lowerCAmelCase					:    List[str]      							= np.array([-0.0_1_9_2, -0.0_2_3_1, -0.0_3_1_8, -0.0_0_5_9, 0.0_0_0_2, -0.0_0_2_0]    )

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



	def 				lowercase     (					self    ):
		lowerCAmelCase					:    int      							= torch_device

		lowerCAmelCase					:    int      							= DanceDiffusionPipeline.from_pretrained('harmonai/maestro-150k' ,					torch_dtype=torch.floataa    )
		lowerCAmelCase					:    Union[str, Any]      							= pipe.to(UpperCamelCase_    )
		pipe.set_progress_bar_config(disable=UpperCamelCase_    )

		lowerCAmelCase					:    Union[str, Any]      							= torch.manual_seed(0    )
		lowerCAmelCase					:    List[str]      							= pipe(generator=UpperCamelCase_ ,					num_inference_steps=100 ,					audio_length_in_s=4.0_9_6    )
		lowerCAmelCase					:    Any      							= output.audios

		lowerCAmelCase					:    Any      							= audio[0, -3:, -3:]

		assert audio.shape == (1, 2, pipe.unet.sample_size)
		lowerCAmelCase					:    Dict      							= np.array([-0.0_3_6_7, -0.0_4_8_8, -0.0_7_7_1, -0.0_5_2_5, -0.0_4_4_4, -0.0_3_4_1]    )

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




'''simple docstring'''




from typing import TYPE_CHECKING

from ...utils import OptionalDependencyNotAvailable, _LazyModule, is_tokenizers_available, is_torch_available


_lowerCAmelCase  :		int		     =	{
    'configuration_nezha': ['NEZHA_PRETRAINED_CONFIG_ARCHIVE_MAP', 'NezhaConfig'],
}

try:
				if not is_torch_available():
								raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
				pass
else:
				_lowerCAmelCase  :		Tuple		     =	[
				    'NEZHA_PRETRAINED_MODEL_ARCHIVE_LIST',
				    'NezhaForNextSentencePrediction',
				    'NezhaForMaskedLM',
				    'NezhaForPreTraining',
				    'NezhaForMultipleChoice',
				    'NezhaForQuestionAnswering',
				    'NezhaForSequenceClassification',
				    'NezhaForTokenClassification',
				    'NezhaModel',
				    'NezhaPreTrainedModel',
				]


if TYPE_CHECKING:
				from .configuration_nezha import NEZHA_PRETRAINED_CONFIG_ARCHIVE_MAP, NezhaConfig

				try:
								if not is_torch_available():
												raise OptionalDependencyNotAvailable()
				except OptionalDependencyNotAvailable:
								pass
				else:
								from .modeling_nezha import (
								    NEZHA_PRETRAINED_MODEL_ARCHIVE_LIST,
								    NezhaForMaskedLM,
								    NezhaForMultipleChoice,
								    NezhaForNextSentencePrediction,
								    NezhaForPreTraining,
								    NezhaForQuestionAnswering,
								    NezhaForSequenceClassification,
								    NezhaForTokenClassification,
								    NezhaModel,
								    NezhaPreTrainedModel,
								)


else:
				import sys

				_lowerCAmelCase  :		str		     =	_LazyModule(__name__, globals()['__file__'], _import_structure, module_spec=__spec__)
