





'''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_		(		__magic_name__						,unittest.TestCase		):
 __lowerCamelCase						:							List[Any]         =					DanceDiffusionPipeline
 __lowerCamelCase						:							Optional[int]         =					UNCONDITIONAL_AUDIO_GENERATION_PARAMS
 __lowerCamelCase						:							Tuple         =					PipelineTesterMixin.required_optional_params - {
     "callback",
     "latents",
     "callback_steps",
     "output_type",
     "num_images_per_prompt",
 }
 __lowerCamelCase						:							List[str]         =					UNCONDITIONAL_AUDIO_GENERATION_BATCH_PARAMS
 __lowerCamelCase						:							Optional[int]         =					False
 __lowerCamelCase						:							int         =					False




 def 							_snake_case   (	self		)    ->							int:
     torch.manual_seed(0		)
     _lowerCAmelCase				=      UNetaDModel(
         block_out_channels=(32, 32, 64)			,     extra_in_channels=16			,     sample_size=512			,     sample_rate=16000			,     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		)    ->							int:
     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		)    ->							Tuple:
     _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.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 							_snake_case   (	self		)    ->							Union[str, Any]:
     return super().test_save_load_local()




 @skip_mps
 def 							_snake_case   (	self		)    ->							List[Any]:
     return super().test_dict_tuple_outputs_equivalent(expected_max_difference=3E-3		)




 @skip_mps
 def 							_snake_case   (	self		)    ->							Union[str, Any]:
     return super().test_save_load_optional_components()




 @skip_mps
 def 							_snake_case   (	self		)    ->							Any:
     return super().test_attention_slicing_forward_pass()





 def 							_snake_case   (	self		)    ->							str:
     super().test_inference_batch_single_identical(expected_max_diff=3E-3		)







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




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




 def 							_snake_case   (	self		)    ->							Any:
     _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=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 							_snake_case   (	self		)    ->							Dict:
     _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=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











'''simple docstring'''


from typing import List, Union

import numpy as np

from ..utils import add_end_docstrings, is_torch_available, is_vision_available, logging, requires_backends
from .base import PIPELINE_INIT_ARGS, Pipeline


if is_vision_available():
       from PIL import Image

       from ..image_utils import load_image

if is_torch_available():
       import torch

       from ..models.auto.modeling_auto import MODEL_FOR_DEPTH_ESTIMATION_MAPPING

_SCREAMING_SNAKE_CASE	 =							logging.get_logger(__name__)

@add_end_docstrings(__magic_name__		)
class    lowerCAmelCase_		(		__magic_name__		):




 def __init__(	self			,     *_lowerCAmelCase			,     **_lowerCAmelCase		)    ->							List[Any]:
     super().__init__(*_lowerCAmelCase			,     **_lowerCAmelCase		)
     requires_backends(self			,     "vision"		)
     self.check_model_type(_lowerCAmelCase		)




 def __call__(	self			,     _lowerCAmelCase			,     **_lowerCAmelCase		)    ->							str:
     return super().__call__(_lowerCAmelCase			,     **_lowerCAmelCase		)




 def 							_snake_case   (	self			,     **_lowerCAmelCase		)    ->							Tuple:
     return {}, {}, {}




 def 							_snake_case   (	self			,     _lowerCAmelCase		)    ->							Union[str, Any]:
     _lowerCAmelCase				=      load_image(_lowerCAmelCase		)
     _lowerCAmelCase				=      image.size
     _lowerCAmelCase				=      self.image_processor(images=_lowerCAmelCase			,     return_tensors=self.framework		)
     return model_inputs




 def 							_snake_case   (	self			,     _lowerCAmelCase		)    ->							Tuple:
     _lowerCAmelCase				=      self.model(**_lowerCAmelCase		)
     return model_outputs





 def 							_snake_case   (	self			,     _lowerCAmelCase		)    ->							Optional[Any]:
     _lowerCAmelCase				=      model_outputs.predicted_depth
     _lowerCAmelCase				=      torch.nn.functional.interpolate(
         predicted_depth.unsqueeze(1		)			,     size=self.image_size[::-1]			,     mode="bicubic"			,     align_corners=_lowerCAmelCase		)
     _lowerCAmelCase				=      prediction.squeeze().cpu().numpy()
     _lowerCAmelCase				=      (output * 255 / np.max(_lowerCAmelCase		)).astype("uint8"		)
     _lowerCAmelCase				=      Image.fromarray(_lowerCAmelCase		)
     _lowerCAmelCase				=      {}
     _lowerCAmelCase				=      predicted_depth
     _lowerCAmelCase				=      depth
     return output_dict


