
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 							_UpperCamelCase					(       __SCREAMING_SNAKE_CASE  ,							unittest.TestCase	):






			'''simple docstring'''

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

			def    lowerCAmelCase__						(							self			:							Tuple   ):
							torch.manual_seed(0   )
							UpperCamelCase_:				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=_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""")    ,						)
							UpperCamelCase_:				Tuple						=    IPNDMScheduler()

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

			def    lowerCAmelCase__						(							self			:							Union[str, Any]    ,						snake_case_			:							Optional[int]    ,						snake_case_			:							Union[str, Any]=0   ):







							if str(_A   ).startswith("""mps"""   ):
											UpperCamelCase_:				Tuple						=    torch.manual_seed(_A   )
							else:
											UpperCamelCase_:				Union[str, Any]						=    torch.Generator(device=_A   ).manual_seed(_A   )
							UpperCamelCase_:				List[Any]						=    {
							    """batch_size""": 1,
							    """generator""": generator,
							    """num_inference_steps""": 4,
							}
							return inputs

			def    lowerCAmelCase__						(							self			:							Dict   ):
							UpperCamelCase_:				Dict						=    """cpu"""  # ensure determinism for the device-dependent torch.Generator
							UpperCamelCase_:				Optional[int]						=    self.get_dummy_components()
							UpperCamelCase_:				Optional[int]						=    DanceDiffusionPipeline(**_A   )
							UpperCamelCase_:				List[Any]						=    pipe.to(_A   )
							pipe.set_progress_bar_config(disable=_A   )

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

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

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

			@skip_mps
			def    lowerCAmelCase__						(							self			:							List[str]   ):
							return super().test_dict_tuple_outputs_equivalent(expected_max_difference=3e-3   )

			@skip_mps
			def    lowerCAmelCase__						(							self			:							List[str]   ):
							return super().test_save_load_optional_components()

			@skip_mps
			def    lowerCAmelCase__						(							self			:							Union[str, Any]   ):
							return super().test_attention_slicing_forward_pass()




			def    lowerCAmelCase__						(							self			:							Union[str, Any]   ):
							super().test_inference_batch_single_identical(expected_max_diff=3e-3   )




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






			'''simple docstring'''

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

			def    lowerCAmelCase__						(							self			:							Tuple   ):
							UpperCamelCase_:				List[str]						=    torch_device

							UpperCamelCase_:				str						=    DanceDiffusionPipeline.from_pretrained("""harmonai/maestro-150k"""   )
							UpperCamelCase_:				Optional[int]						=    pipe.to(_A   )
							pipe.set_progress_bar_config(disable=_A   )

							UpperCamelCase_:				List[str]						=    torch.manual_seed(0   )
							UpperCamelCase_:				Optional[Any]						=    pipe(generator=_A    ,						num_inference_steps=100    ,						audio_length_in_s=4.096   )
							UpperCamelCase_:				Union[str, Any]						=    output.audios

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

							assert audio.shape == (1, 2, pipe.unet.sample_size)
							UpperCamelCase_:				Any						=    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    lowerCAmelCase__						(							self			:							Any   ):
							UpperCamelCase_:				List[Any]						=    torch_device

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

							UpperCamelCase_:				Any						=    torch.manual_seed(0   )
							UpperCamelCase_:				Union[str, Any]						=    pipe(generator=_A    ,						num_inference_steps=100    ,						audio_length_in_s=4.096   )
							UpperCamelCase_:				int						=    output.audios

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

							assert audio.shape == (1, 2, pipe.unet.sample_size)
							UpperCamelCase_:				Dict						=    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













"""simple docstring"""


from bisect import bisect
from itertools import accumulate

def      UpperCamelCase							(SCREAMING_SNAKE_CASE							,     SCREAMING_SNAKE_CASE							,     SCREAMING_SNAKE_CASE							,     SCREAMING_SNAKE_CASE     ):
						UpperCamelCase				:		Dict				   =			sorted(zip(SCREAMING_SNAKE_CASE							,     SCREAMING_SNAKE_CASE     )							,     key=lambda SCREAMING_SNAKE_CASE     : x[0] / x[1]							,     reverse=SCREAMING_SNAKE_CASE     )
						UpperCamelCase   ,						UpperCamelCase				:		int				   =			[i[0] for i in r], [i[1] for i in r]
						UpperCamelCase				:		Optional[Any]				   =			list(accumulate(SCREAMING_SNAKE_CASE     )     )
						UpperCamelCase				:		Optional[Any]				   =			bisect(SCREAMING_SNAKE_CASE							,     SCREAMING_SNAKE_CASE     )
						return (
						    0
						    if k == 0
						    else sum(vl[:k]     ) + (w - acc[k - 1]) * (vl[k]) / (wt[k])
						    if k != n
						    else sum(vl[:k]     )
						)


if __name__ == "__main__":
		import doctest

		doctest.testmod()



