
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 ):
				_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   __a		(      self  )       ->   Tuple:
							torch.manual_seed(0  )
							lowerCAmelCase_      			=			UNetaDModel(
							    block_out_channels=(32, 32, 64)       ,       extra_in_channels=16       ,       sample_size=512       ,       sample_rate=16_000       ,       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_      			=			IPNDMScheduler()

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




				def   __a		(      self       ,       _UpperCamelCase       ,       _UpperCamelCase=0  )       ->   List[Any]:
							if str(_UpperCamelCase  ).startswith("mps"  ):
										lowerCAmelCase_      			=			torch.manual_seed(_UpperCamelCase  )
							else:
										lowerCAmelCase_      			=			torch.Generator(device=_UpperCamelCase  ).manual_seed(_UpperCamelCase  )
							lowerCAmelCase_      			=			{
							    "batch_size": 1,
							    "generator": generator,
							    "num_inference_steps": 4,
							}
							return inputs




				def   __a		(      self  )       ->   Optional[int]:
							lowerCAmelCase_      			=			"cpu"  # ensure determinism for the device-dependent torch.Generator
							lowerCAmelCase_      			=			self.get_dummy_components()
							lowerCAmelCase_      			=			DanceDiffusionPipeline(**_UpperCamelCase  )
							lowerCAmelCase_      			=			pipe.to(_UpperCamelCase  )
							pipe.set_progress_bar_config(disable=_UpperCamelCase  )

							lowerCAmelCase_      			=			self.get_dummy_inputs(_UpperCamelCase  )
							lowerCAmelCase_      			=			pipe(**_UpperCamelCase  )
							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  )       ->   Optional[Any]:
							return super().test_save_load_local()




				@skip_mps
				def   __a		(      self  )       ->   Optional[int]:
							return super().test_dict_tuple_outputs_equivalent(expected_max_difference=3e-3  )




				@skip_mps
				def   __a		(      self  )       ->   List[str]:
							return super().test_save_load_optional_components()




				@skip_mps
				def   __a		(      self  )       ->   List[Any]:
							return super().test_attention_slicing_forward_pass()




				def   __a		(      self  )       ->   List[str]:
							super().test_inference_batch_single_identical(expected_max_diff=3e-3  )





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




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




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

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

							lowerCAmelCase_      			=			torch.manual_seed(0  )
							lowerCAmelCase_      			=			pipe(generator=_UpperCamelCase       ,       num_inference_steps=100       ,       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  )       ->   Dict:
							lowerCAmelCase_      			=			torch_device

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

							lowerCAmelCase_      			=			torch.manual_seed(0  )
							lowerCAmelCase_      			=			pipe(generator=_UpperCamelCase       ,       num_inference_steps=100       ,       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,
)


_A     =  {
    "configuration_wav2vec2": ["WAV_2_VEC_2_PRETRAINED_CONFIG_ARCHIVE_MAP", "Wav2Vec2Config"],
    "feature_extraction_wav2vec2": ["Wav2Vec2FeatureExtractor"],
    "processing_wav2vec2": ["Wav2Vec2Processor"],
    "tokenization_wav2vec2": ["Wav2Vec2CTCTokenizer", "Wav2Vec2Tokenizer"],
}


try:
			if not is_torch_available():
						raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
			pass
else:
			_A     =  [
			    "WAV_2_VEC_2_PRETRAINED_MODEL_ARCHIVE_LIST",
			    "Wav2Vec2ForAudioFrameClassification",
			    "Wav2Vec2ForCTC",
			    "Wav2Vec2ForMaskedLM",
			    "Wav2Vec2ForPreTraining",
			    "Wav2Vec2ForSequenceClassification",
			    "Wav2Vec2ForXVector",
			    "Wav2Vec2Model",
			    "Wav2Vec2PreTrainedModel",
			]

try:
			if not is_tf_available():
						raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
			pass
else:
			_A     =  [
			    "TF_WAV_2_VEC_2_PRETRAINED_MODEL_ARCHIVE_LIST",
			    "TFWav2Vec2ForCTC",
			    "TFWav2Vec2Model",
			    "TFWav2Vec2PreTrainedModel",
			    "TFWav2Vec2ForSequenceClassification",
			]

try:
			if not is_flax_available():
						raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
			pass
else:
			_A     =  [
			    "FlaxWav2Vec2ForCTC",
			    "FlaxWav2Vec2ForPreTraining",
			    "FlaxWav2Vec2Model",
			    "FlaxWav2Vec2PreTrainedModel",
			]


if TYPE_CHECKING:
			from .configuration_wavaveca import WAV_2_VEC_2_PRETRAINED_CONFIG_ARCHIVE_MAP, WavaVecaConfig
			from .feature_extraction_wavaveca import WavaVecaFeatureExtractor
			from .processing_wavaveca import WavaVecaProcessor
			from .tokenization_wavaveca import WavaVecaCTCTokenizer, WavaVecaTokenizer

			try:
						if not is_torch_available():
									raise OptionalDependencyNotAvailable()
			except OptionalDependencyNotAvailable:
						pass
			else:
						from .modeling_wavaveca import (
						    WAV_2_VEC_2_PRETRAINED_MODEL_ARCHIVE_LIST,
						    WavaVecaForAudioFrameClassification,
						    WavaVecaForCTC,
						    WavaVecaForMaskedLM,
						    WavaVecaForPreTraining,
						    WavaVecaForSequenceClassification,
						    WavaVecaForXVector,
						    WavaVecaModel,
						    WavaVecaPreTrainedModel,
						)

			try:
						if not is_tf_available():
									raise OptionalDependencyNotAvailable()
			except OptionalDependencyNotAvailable:
						pass
			else:
						from .modeling_tf_wavaveca import (
						    TF_WAV_2_VEC_2_PRETRAINED_MODEL_ARCHIVE_LIST,
						    TFWavaVecaForCTC,
						    TFWavaVecaForSequenceClassification,
						    TFWavaVecaModel,
						    TFWavaVecaPreTrainedModel,
						)

			try:
						if not is_flax_available():
									raise OptionalDependencyNotAvailable()
			except OptionalDependencyNotAvailable:
						pass
			else:
						from .modeling_tf_wavaveca import (
						    FlaxWavaVecaForCTC,
						    FlaxWavaVecaForPreTraining,
						    FlaxWavaVecaModel,
						    FlaxWavaVecaPreTrainedModel,
						)


else:
			import sys

			_A     =  _LazyModule(__name__, globals()["__file__"], _import_structure, module_spec=__spec__)



