



"""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    ):





					'''simple docstring'''



					lowercase__   :       Tuple										= DanceDiffusionPipeline
					lowercase__   :       str										= UNCONDITIONAL_AUDIO_GENERATION_PARAMS
					lowercase__   :       int										= PipelineTesterMixin.required_optional_params - {
					    'callback',
					    'latents',
					    'callback_steps',
					    'output_type',
					    'num_images_per_prompt',
					}
					lowercase__   :       List[Any]										= UNCONDITIONAL_AUDIO_GENERATION_BATCH_PARAMS
					lowercase__   :       str										= False
					lowercase__   :       int										= False

					def 			snake_case__      (    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=lowerCamelCase__				,       use_timestep_embedding=lowerCamelCase__				,       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 			snake_case__      (    self				,       lowerCamelCase__				,       lowerCamelCase__=0   ):
									if str(lowerCamelCase__   ).startswith('''mps'''   ):
													_lowerCamelCase        =				torch.manual_seed(lowerCamelCase__   )
									else:
													_lowerCamelCase        =				torch.Generator(device=lowerCamelCase__   ).manual_seed(lowerCamelCase__   )
									_lowerCamelCase        =				{
									    '''batch_size''': 1,
									    '''generator''': generator,
									    '''num_inference_steps''': 4,
									}
									return inputs

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

									_lowerCamelCase        =				self.get_dummy_inputs(lowerCamelCase__   )
									_lowerCamelCase        =				pipe(**lowerCamelCase__   )
									_lowerCamelCase        =				output.audios

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

									assert audio.shape == (1, 2, components["unet"].sample_size)
									_lowerCamelCase        =				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 			snake_case__      (    self   ):
									return super().test_save_load_local()

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

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

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





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


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





					'''simple docstring'''



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

					def 			snake_case__      (    self   ):
									_lowerCamelCase        =				torch_device

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

									_lowerCamelCase        =				torch.manual_seed(0   )
									_lowerCamelCase        =				pipe(generator=lowerCamelCase__				,       num_inference_steps=1_0_0				,       audio_length_in_s=4.0_9_6   )
									_lowerCamelCase        =				output.audios

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

									assert audio.shape == (1, 2, pipe.unet.sample_size)
									_lowerCamelCase        =				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 			snake_case__      (    self   ):
									_lowerCamelCase        =				torch_device

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

									_lowerCamelCase        =				torch.manual_seed(0   )
									_lowerCamelCase        =				pipe(generator=lowerCamelCase__				,       num_inference_steps=1_0_0				,       audio_length_in_s=4.0_9_6   )
									_lowerCamelCase        =				output.audios

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

									assert audio.shape == (1, 2, pipe.unet.sample_size)
									_lowerCamelCase        =				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_flax_available,
    is_tf_available,
    is_torch_available,
)


__SCREAMING_SNAKE_CASE  :						Optional[Any]    							=      {
    '''configuration_vision_encoder_decoder''': ['''VisionEncoderDecoderConfig''', '''VisionEncoderDecoderOnnxConfig''']
}

try:
			if not is_torch_available():
						raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
			pass
else:
			__SCREAMING_SNAKE_CASE  :						str    							=      ['''VisionEncoderDecoderModel''']

try:
			if not is_tf_available():
						raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
			pass
else:
			__SCREAMING_SNAKE_CASE  :						Dict    							=      ['''TFVisionEncoderDecoderModel''']

try:
			if not is_flax_available():
						raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
			pass
else:
			__SCREAMING_SNAKE_CASE  :						List[Any]    							=      ['''FlaxVisionEncoderDecoderModel''']

if TYPE_CHECKING:
			from .configuration_vision_encoder_decoder import VisionEncoderDecoderConfig, VisionEncoderDecoderOnnxConfig

			try:
						if not is_torch_available():
									raise OptionalDependencyNotAvailable()
			except OptionalDependencyNotAvailable:
						pass
			else:
						from .modeling_vision_encoder_decoder import VisionEncoderDecoderModel

			try:
						if not is_tf_available():
									raise OptionalDependencyNotAvailable()
			except OptionalDependencyNotAvailable:
						pass
			else:
						from .modeling_tf_vision_encoder_decoder import TFVisionEncoderDecoderModel

			try:
						if not is_flax_available():
									raise OptionalDependencyNotAvailable()
			except OptionalDependencyNotAvailable:
						pass
			else:
						from .modeling_flax_vision_encoder_decoder import FlaxVisionEncoderDecoderModel

else:
			import sys

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