






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	):
 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 		UpperCAmelCase  (							self       )->					int:







       '''simple docstring'''





       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 		UpperCAmelCase  (							self		,    __UpperCAmelCase		,    __UpperCAmelCase=0       )->					Tuple:







       '''simple docstring'''





       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 		UpperCAmelCase  (							self       )->					List[Any]:







       '''simple docstring'''





       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_265, 1.0_000, -0.8_388, 0.1_175, 0.9_498, -1.0_000]       )
       assert np.abs(audio_slice.flatten() - expected_slice       ).max() < 1E-2

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







       '''simple docstring'''





       return super().test_save_load_local()

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







       '''simple docstring'''





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

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







       '''simple docstring'''





       return super().test_save_load_optional_components()

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







       '''simple docstring'''





       return super().test_attention_slicing_forward_pass()


 def 		UpperCAmelCase  (							self       )->					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 		UpperCAmelCase  (							self       )->					Any:







       '''simple docstring'''





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

 def 		UpperCAmelCase  (							self       )->					List[Any]:







       '''simple docstring'''





       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.0_192, -0.0_231, -0.0_318, -0.0_059, 0.0_002, -0.0_020]       )

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


 def 		UpperCAmelCase  (							self       )->					Any:







       '''simple docstring'''





       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.0_367, -0.0_488, -0.0_771, -0.0_525, -0.0_444, -0.0_341]       )

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














from typing import Dict, List, Optional, Tuple, Union

import torch

from ...models import AutoencoderKL, TransformeraDModel
from ...schedulers import KarrasDiffusionSchedulers
from ...utils import randn_tensor
from ..pipeline_utils import DiffusionPipeline, ImagePipelineOutput


class      lowercase__     (			_UpperCAmelCase	):

	def __init__(							self		,    __UpperCAmelCase		,    __UpperCAmelCase		,    __UpperCAmelCase		,    __UpperCAmelCase = None		,    )->					List[str]:







							'''simple docstring'''





							super().__init__()
							self.register_modules(transformer=__UpperCAmelCase		,    vae=__UpperCAmelCase		,    scheduler=__UpperCAmelCase       )

							# create a imagenet -> id dictionary for easier use
							lowerCAmelCase__       =						{}
							if idalabel is not None:
													for key, value in idalabel.items():
																			for label in value.split(","       ):
																									lowerCAmelCase__       =						int(__UpperCAmelCase       )
													lowerCAmelCase__       =						dict(sorted(self.labels.items()       )       )

	def 		UpperCAmelCase  (							self		,    __UpperCAmelCase       )->					List[int]:







							'''simple docstring'''





							if not isinstance(__UpperCAmelCase		,    __UpperCAmelCase       ):
													lowerCAmelCase__       =						list(__UpperCAmelCase       )

							for l in label:
													if l not in self.labels:
																			raise ValueError(
																			    F"{l} does not exist. Please make sure to select one of the following labels: \n {self.labels}."       )

							return [self.labels[l] for l in label]


	@torch.no_grad()
	def __call__(							self		,    __UpperCAmelCase		,    __UpperCAmelCase = 4.0		,    __UpperCAmelCase = None		,    __UpperCAmelCase = 50		,    __UpperCAmelCase = "pil"		,    __UpperCAmelCase = True		,    )->					Union[ImagePipelineOutput, Tuple]:







							'''simple docstring'''





							lowerCAmelCase__       =						len(__UpperCAmelCase       )
							lowerCAmelCase__       =						self.transformer.config.sample_size
							lowerCAmelCase__       =						self.transformer.config.in_channels

							lowerCAmelCase__       =						randn_tensor(
							    shape=(batch_size, latent_channels, latent_size, latent_size)		,    generator=__UpperCAmelCase		,    device=self.device		,    dtype=self.transformer.dtype		,    )
							lowerCAmelCase__       =						torch.cat([latents] * 2       ) if guidance_scale > 1 else latents

							lowerCAmelCase__       =						torch.tensor(__UpperCAmelCase		,    device=self.device       ).reshape(-1       )
							lowerCAmelCase__       =						torch.tensor([1000] * batch_size		,    device=self.device       )
							lowerCAmelCase__       =						torch.cat([class_labels, class_null]		,    0       ) if guidance_scale > 1 else class_labels

							# set step values
							self.scheduler.set_timesteps(__UpperCAmelCase       )

							for t in self.progress_bar(self.scheduler.timesteps       ):
													if guidance_scale > 1:
																			lowerCAmelCase__       =						latent_model_input[: len(__UpperCAmelCase       ) // 2]
																			lowerCAmelCase__       =						torch.cat([half, half]		,    dim=0       )
													lowerCAmelCase__       =						self.scheduler.scale_model_input(__UpperCAmelCase		,    __UpperCAmelCase       )

													lowerCAmelCase__       =						t
													if not torch.is_tensor(__UpperCAmelCase       ):
																			# TODO: this requires sync between CPU and GPU. So try to pass timesteps as tensors if you can
																			# This would be a good case for the `match` statement (Python 3.10+)
																			lowerCAmelCase__       =						latent_model_input.device.type == "mps"
																			if isinstance(__UpperCAmelCase		,    __UpperCAmelCase       ):
																									lowerCAmelCase__       =						torch.floataa if is_mps else torch.floataa
																			else:
																									lowerCAmelCase__       =						torch.intaa if is_mps else torch.intaa
																			lowerCAmelCase__       =						torch.tensor([timesteps]		,    dtype=__UpperCAmelCase		,    device=latent_model_input.device       )
													elif len(timesteps.shape       ) == 0:
																			lowerCAmelCase__       =						timesteps[None].to(latent_model_input.device       )
													# broadcast to batch dimension in a way that's compatible with ONNX/Core ML
													lowerCAmelCase__       =						timesteps.expand(latent_model_input.shape[0]       )
													# predict noise model_output
													lowerCAmelCase__       =						self.transformer(
													    __UpperCAmelCase		,    timestep=__UpperCAmelCase		,    class_labels=__UpperCAmelCase       ).sample

													# perform guidance
													if guidance_scale > 1:
																			lowerCAmelCase__      ,							lowerCAmelCase__       =						noise_pred[:, :latent_channels], noise_pred[:, latent_channels:]
																			lowerCAmelCase__      ,							lowerCAmelCase__       =						torch.split(__UpperCAmelCase		,    len(__UpperCAmelCase       ) // 2		,    dim=0       )

																			lowerCAmelCase__       =						uncond_eps + guidance_scale * (cond_eps - uncond_eps)
																			lowerCAmelCase__       =						torch.cat([half_eps, half_eps]		,    dim=0       )

																			lowerCAmelCase__       =						torch.cat([eps, rest]		,    dim=1       )

													# learned sigma
													if self.transformer.config.out_channels // 2 == latent_channels:
																			lowerCAmelCase__      ,							lowerCAmelCase__       =						torch.split(__UpperCAmelCase		,    __UpperCAmelCase		,    dim=1       )
													else:
																			lowerCAmelCase__       =						noise_pred

													# compute previous image: x_t -> x_t-1
													lowerCAmelCase__       =						self.scheduler.step(__UpperCAmelCase		,    __UpperCAmelCase		,    __UpperCAmelCase       ).prev_sample

							if guidance_scale > 1:
													lowerCAmelCase__      ,							lowerCAmelCase__       =						latent_model_input.chunk(2		,    dim=0       )
							else:
													lowerCAmelCase__       =						latent_model_input

							lowerCAmelCase__       =						1 / self.vae.config.scaling_factor * latents
							lowerCAmelCase__       =						self.vae.decode(__UpperCAmelCase       ).sample

							lowerCAmelCase__       =						(samples / 2 + 0.5).clamp(0		,    1       )

							# we always cast to float32 as this does not cause significant overhead and is compatible with bfloat16
							lowerCAmelCase__       =						samples.cpu().permute(0		,    2		,    3		,    1       ).float().numpy()

							if output_type == "pil":
													lowerCAmelCase__       =						self.numpy_to_pil(__UpperCAmelCase       )

							if not return_dict:
													return (samples,)

							return ImagePipelineOutput(images=__UpperCAmelCase       )






