




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				(							lowerCAmelCase					,       unittest.TestCase):
       _a           =  DanceDiffusionPipeline
       _a           =  UNCONDITIONAL_AUDIO_GENERATION_PARAMS
       _a           =  PipelineTesterMixin.required_optional_params - {
           '''callback''',
           '''latents''',
           '''callback_steps''',
           '''output_type''',
           '''num_images_per_prompt''',
       }
       _a           =  UNCONDITIONAL_AUDIO_GENERATION_BATCH_PARAMS
       _a           =  False
       _a           =  False






       def     SCREAMING_SNAKE_CASE			(      self:    Optional[int] ):
         torch.manual_seed(0 )
         lowercase			:str							=  UNetaDModel(
             block_out_channels=(32, 32, 64)				,    extra_in_channels=16				,    sample_size=5_12				,    sample_rate=1_60_00				,    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")				,    )
         lowercase			:List[Any]							=  IPNDMScheduler()

         lowercase			:Optional[int]							=  {
             "unet": unet,
             "scheduler": scheduler,
         }
         return components






       def     SCREAMING_SNAKE_CASE			(      self:    List[Any]				,    _lowerCAmelCase:    Union[str, Any]				,    _lowerCAmelCase:    str=0 ):
         if str(_lowerCAmelCase ).startswith("mps" ):
           lowercase			:Optional[Any]							=  torch.manual_seed(_lowerCAmelCase )
         else:
           lowercase			:Union[str, Any]							=  torch.Generator(device=_lowerCAmelCase ).manual_seed(_lowerCAmelCase )
         lowercase			:Any							=  {
             "batch_size": 1,
             "generator": generator,
             "num_inference_steps": 4,
         }
         return inputs






       def     SCREAMING_SNAKE_CASE			(      self:    str ):
         lowercase			:List[str]							=  "cpu"  # ensure determinism for the device-dependent torch.Generator
         lowercase			:List[Any]							=  self.get_dummy_components()
         lowercase			:int							=  DanceDiffusionPipeline(**_lowerCAmelCase )
         lowercase			:Tuple							=  pipe.to(_lowerCAmelCase )
         pipe.set_progress_bar_config(disable=_lowerCAmelCase )

         lowercase			:Dict							=  self.get_dummy_inputs(_lowerCAmelCase )
         lowercase			:Dict							=  pipe(**_lowerCAmelCase )
         lowercase			:Optional[int]							=  output.audios

         lowercase			:Tuple							=  audio[0, -3:, -3:]

         assert audio.shape == (1, 2, components["unet"].sample_size)
         lowercase			:Optional[int]							=  np.array([-0.72_65, 1.00_00, -0.83_88, 0.11_75, 0.94_98, -1.00_00] )
         assert np.abs(audio_slice.flatten() - expected_slice ).max() < 1e-2






       @skip_mps
       def     SCREAMING_SNAKE_CASE			(      self:    Any ):
         return super().test_save_load_local()






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






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






       @skip_mps
       def     SCREAMING_SNAKE_CASE			(      self:    Any ):
         return super().test_attention_slicing_forward_pass()






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







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






       def     SCREAMING_SNAKE_CASE			(      self:    Optional[Any] ):
         # clean up the VRAM after each test
         super().tearDown()
         gc.collect()
         torch.cuda.empty_cache()






       def     SCREAMING_SNAKE_CASE			(      self:    List[Any] ):
         lowercase			:List[Any]							=  torch_device

         lowercase			:List[Any]							=  DanceDiffusionPipeline.from_pretrained("harmonai/maestro-150k" )
         lowercase			:Tuple							=  pipe.to(_lowerCAmelCase )
         pipe.set_progress_bar_config(disable=_lowerCAmelCase )

         lowercase			:List[Any]							=  torch.manual_seed(0 )
         lowercase			:str							=  pipe(generator=_lowerCAmelCase				,    num_inference_steps=1_00				,    audio_length_in_s=4.0_96 )
         lowercase			:Union[str, Any]							=  output.audios

         lowercase			:Optional[Any]							=  audio[0, -3:, -3:]

         assert audio.shape == (1, 2, pipe.unet.sample_size)
         lowercase			:List[str]							=  np.array([-0.01_92, -0.02_31, -0.03_18, -0.00_59, 0.00_02, -0.00_20] )

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






       def     SCREAMING_SNAKE_CASE			(      self:    int ):
         lowercase			:Optional[Any]							=  torch_device

         lowercase			:Any							=  DanceDiffusionPipeline.from_pretrained("harmonai/maestro-150k"				,    torch_dtype=torch.floataa )
         lowercase			:Dict							=  pipe.to(_lowerCAmelCase )
         pipe.set_progress_bar_config(disable=_lowerCAmelCase )

         lowercase			:int							=  torch.manual_seed(0 )
         lowercase			:Dict							=  pipe(generator=_lowerCAmelCase				,    num_inference_steps=1_00				,    audio_length_in_s=4.0_96 )
         lowercase			:Dict							=  output.audios

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

         assert audio.shape == (1, 2, pipe.unet.sample_size)
         lowercase			:Union[str, Any]							=  np.array([-0.03_67, -0.04_88, -0.07_71, -0.05_25, -0.04_44, -0.03_41] )

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









import os
from math import logaa



def 					UpperCAmelCase__     (   lowerCamelCase = "base_exp.txt"      ):
  lowercase			:float							=  0
  lowercase			:str							=  0
  for i, line in enumerate(open(os.path.join(os.path.dirname(lowerCamelCase      ),	lowerCamelCase      )      )      ):
    lowercase		,			lowercase			:str							=  list(map(lowerCamelCase,	line.split(","      )      )      )
    if x * logaa(lowerCamelCase      ) > largest:
      lowercase			:Optional[Any]							=  x * logaa(lowerCamelCase      )
      lowercase			:Any							=  i + 1
  return result


if __name__ == "__main__":
     print(solution())



