



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 		_lowercase		(       UpperCAmelCase__,   unittest.TestCase	):
   _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			:       Union[str, Any]							)       ->				int:


    """simple docstring"""






    torch.manual_seed(0							)
    a						=		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=__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")							,   )
    a						=		IPNDMScheduler()

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

   def 		A							(      self			:       Dict							,   __lowerCAmelCase			:       int							,   __lowerCAmelCase			:       Optional[int]=0							)       ->				Tuple:


    """simple docstring"""






    if str(__lowerCAmelCase							).startswith("mps"							):
     a						=		torch.manual_seed(__lowerCAmelCase							)
    else:
     a						=		torch.Generator(device=__lowerCAmelCase							).manual_seed(__lowerCAmelCase							)
    a						=		{
        "batch_size": 1,
        "generator": generator,
        "num_inference_steps": 4,
    }
    return inputs

   def 		A							(      self			:       Tuple							)       ->				List[Any]:


    """simple docstring"""






    a						=		"cpu"  # ensure determinism for the device-dependent torch.Generator
    a						=		self.get_dummy_components()
    a						=		DanceDiffusionPipeline(**__lowerCAmelCase							)
    a						=		pipe.to(__lowerCAmelCase							)
    pipe.set_progress_bar_config(disable=__lowerCAmelCase							)

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

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

    assert audio.shape == (1, 2, components["unet"].sample_size)
    a						=		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 		A							(      self			:       Dict							)       ->				List[str]:


    """simple docstring"""






    return super().test_save_load_local()

   @skip_mps
   def 		A							(      self			:       int							)       ->				List[str]:


    """simple docstring"""






    return super().test_dict_tuple_outputs_equivalent(expected_max_difference=3E-3							)

   @skip_mps
   def 		A							(      self			:       Optional[Any]							)       ->				Union[str, Any]:


    """simple docstring"""






    return super().test_save_load_optional_components()

   @skip_mps
   def 		A							(      self			:       List[Any]							)       ->				List[Any]:


    """simple docstring"""






    return super().test_attention_slicing_forward_pass()




   def 		A							(      self			:       Tuple							)       ->				Union[str, Any]:


    """simple docstring"""






    super().test_inference_batch_single_identical(expected_max_diff=3E-3							)



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

   def 		A							(      self			:       Any							)       ->				Any:


    """simple docstring"""






    super().tearDown()
    gc.collect()
    torch.cuda.empty_cache()

   def 		A							(      self			:       Tuple							)       ->				List[Any]:


    """simple docstring"""






    a						=		torch_device

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

    a						=		torch.manual_seed(0							)
    a						=		pipe(generator=__lowerCAmelCase							,   num_inference_steps=100							,   audio_length_in_s=4.0_9_6							)
    a						=		output.audios

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

    assert audio.shape == (1, 2, pipe.unet.sample_size)
    a						=		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 		A							(      self			:       Optional[Any]							)       ->				Any:


    """simple docstring"""






    a						=		torch_device

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

    a						=		torch.manual_seed(0							)
    a						=		pipe(generator=__lowerCAmelCase							,   num_inference_steps=100							,   audio_length_in_s=4.0_9_6							)
    a						=		output.audios

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

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






def 	UpperCAmelCase__      (      UpperCAmelCase__ :int ,					UpperCAmelCase__ :int					):






 '''simple docstring'''

 if a < 0 or b < 0:
  raise ValueError("the value of both inputs must be positive"					)

 a						=		str(bin(UpperCAmelCase__					)					)[2:]  # remove the leading "0b"
 a						=		str(bin(UpperCAmelCase__					)					)[2:]  # remove the leading "0b"

 a						=		max(len(UpperCAmelCase__					) ,					len(UpperCAmelCase__					)					)

 return "0b" + "".join(
     str(int(char_a == "1" and char_b == "1"					)					)
     for char_a, char_b in zip(a_binary.zfill(UpperCAmelCase__					) ,					b_binary.zfill(UpperCAmelCase__					)					)					)


if __name__ == "__main__":
     import doctest

     doctest.testmod()

