






import gc
import unittest

import numpy as np
import torch

from diffusers import (
    AudioDiffusionPipeline,
    AutoencoderKL,
    DDIMScheduler,
    DDPMScheduler,
    DiffusionPipeline,
    Mel,
    UNetaDConditionModel,
    UNetaDModel,
)
from diffusers.utils import slow, torch_device
from diffusers.utils.testing_utils import enable_full_determinism, require_torch_gpu


enable_full_determinism()




class 							lowerCAmelCase__  (	unittest.TestCase						):







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







					@property
					def  _lowercase			(       self      :     Dict):
											torch.manual_seed(0)
											A__	:							Union[str, Any]						    = UNetaDModel(
											    sample_size=(32, 64)			,		in_channels=1			,		out_channels=1			,		layers_per_block=2			,		block_out_channels=(128, 128)			,		down_block_types=("AttnDownBlock2D", "DownBlock2D")			,		up_block_types=("UpBlock2D", "AttnUpBlock2D")			,		)
											return model







					@property
					def  _lowercase			(       self      :     Tuple):
											torch.manual_seed(0)
											A__	:							List[str]						    = UNetaDConditionModel(
											    sample_size=(64, 32)			,		in_channels=1			,		out_channels=1			,		layers_per_block=2			,		block_out_channels=(128, 128)			,		down_block_types=("CrossAttnDownBlock2D", "DownBlock2D")			,		up_block_types=("UpBlock2D", "CrossAttnUpBlock2D")			,		cross_attention_dim=10			,		)
											return model







					@property
					def  _lowercase			(       self      :     Optional[int]):
											torch.manual_seed(0)
											A__	:							Union[str, Any]						    = AutoencoderKL(
											    sample_size=(128, 64)			,		in_channels=1			,		out_channels=1			,		latent_channels=1			,		layers_per_block=2			,		block_out_channels=(128, 128)			,		down_block_types=("DownEncoderBlock2D", "DownEncoderBlock2D")			,		up_block_types=("UpDecoderBlock2D", "UpDecoderBlock2D")			,		)
											A__	:							str						    = UNetaDModel(
											    sample_size=(64, 32)			,		in_channels=1			,		out_channels=1			,		layers_per_block=2			,		block_out_channels=(128, 128)			,		down_block_types=("AttnDownBlock2D", "DownBlock2D")			,		up_block_types=("UpBlock2D", "AttnUpBlock2D")			,		)
											return vqvae, unet







					@slow
					def  _lowercase			(       self      :     Union[str, Any]):
											A__	:							Dict						    = "cpu"  # ensure determinism for the device-dependent torch.Generator
											A__	:							Union[str, Any]						    = Mel(
											    x_res=self.dummy_unet.config.sample_size[1]			,		y_res=self.dummy_unet.config.sample_size[0]			,		)

											A__	:							Dict						    = DDPMScheduler()
											A__	:							Optional[Any]						    = AudioDiffusionPipeline(vqvae=_A			,		unet=self.dummy_unet			,		mel=_A			,		scheduler=_A)
											A__	:							Union[str, Any]						    = pipe.to(_A)
											pipe.set_progress_bar_config(disable=_A)

											A__	:							int						    = torch.Generator(device=_A).manual_seed(42)
											A__	:							str						    = pipe(generator=_A			,		steps=4)
											A__	:							Optional[int]						    = output.audios[0]
											A__	:							Union[str, Any]						    = output.images[0]

											A__	:							Dict						    = torch.Generator(device=_A).manual_seed(42)
											A__	:							Dict						    = pipe(generator=_A			,		steps=4			,		return_dict=_A)
											A__	:							Optional[Any]						    = output[0][0]

											assert audio.shape == (1, (self.dummy_unet.config.sample_size[1] - 1) * mel.hop_length)
											assert (
											    image.height == self.dummy_unet.config.sample_size[0]
											    and image.width == self.dummy_unet.config.sample_size[1]
											)
											A__	:							Any						    = np.frombuffer(image.tobytes()			,		dtype="uint8")[:10]
											A__	:							List[str]						    = np.frombuffer(image_from_tuple.tobytes()			,		dtype="uint8")[:10]
											A__	:							Optional[Any]						    = np.array([69, 255, 255, 255, 0, 0, 77, 181, 12, 127])

											assert np.abs(image_slice.flatten() - expected_slice).max() == 0
											assert np.abs(image_from_tuple_slice.flatten() - expected_slice).max() == 0

											A__	:							str						    = Mel(
											    x_res=self.dummy_vqvae_and_unet[0].config.sample_size[1]			,		y_res=self.dummy_vqvae_and_unet[0].config.sample_size[0]			,		)

											A__	:							Union[str, Any]						    = DDIMScheduler()
											A__	:							Any						    = self.dummy_vqvae_and_unet
											A__	:							Optional[int]						    = AudioDiffusionPipeline(
											    vqvae=self.dummy_vqvae_and_unet[0]			,		unet=dummy_vqvae_and_unet[1]			,		mel=_A			,		scheduler=_A)
											A__	:							int						    = pipe.to(_A)
											pipe.set_progress_bar_config(disable=_A)

											np.random.seed(0)
											A__	:							Optional[Any]						    = np.random.uniform(-1			,		1			,		((dummy_vqvae_and_unet[0].config.sample_size[1] - 1) * mel.hop_length,))
											A__	:							Optional[int]						    = torch.Generator(device=_A).manual_seed(42)
											A__	:							int						    = pipe(raw_audio=_A			,		generator=_A			,		start_step=5			,		steps=10)
											A__	:							List[str]						    = output.images[0]

											assert (
											    image.height == self.dummy_vqvae_and_unet[0].config.sample_size[0]
											    and image.width == self.dummy_vqvae_and_unet[0].config.sample_size[1]
											)
											A__	:							Optional[Any]						    = np.frombuffer(image.tobytes()			,		dtype="uint8")[:10]
											A__	:							Tuple						    = np.array([120, 117, 110, 109, 138, 167, 138, 148, 132, 121])

											assert np.abs(image_slice.flatten() - expected_slice).max() == 0

											A__	:							str						    = self.dummy_unet_condition
											A__	:							List[Any]						    = AudioDiffusionPipeline(
											    vqvae=self.dummy_vqvae_and_unet[0]			,		unet=_A			,		mel=_A			,		scheduler=_A)
											A__	:							Optional[Any]						    = pipe.to(_A)
											pipe.set_progress_bar_config(disable=_A)

											np.random.seed(0)
											A__	:							Tuple						    = torch.rand((1, 1, 10))
											A__	:							Optional[Any]						    = pipe(generator=_A			,		encoding=_A)
											A__	:							Dict						    = output.images[0]
											A__	:							Tuple						    = np.frombuffer(image.tobytes()			,		dtype="uint8")[:10]
											A__	:							Any						    = np.array([107, 103, 120, 127, 142, 122, 113, 122, 97, 111])

											assert np.abs(image_slice.flatten() - expected_slice).max() == 0




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







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







					def  _lowercase			(       self      :     Dict):
											A__	:							Optional[int]						    = torch_device

											A__	:							Optional[Any]						    = DiffusionPipeline.from_pretrained("teticio/audio-diffusion-ddim-256")
											A__	:							Optional[int]						    = pipe.to(_A)
											pipe.set_progress_bar_config(disable=_A)

											A__	:							List[Any]						    = torch.Generator(device=_A).manual_seed(42)
											A__	:							Tuple						    = pipe(generator=_A)
											A__	:							Tuple						    = output.audios[0]
											A__	:							str						    = output.images[0]

											assert audio.shape == (1, (pipe.unet.config.sample_size[1] - 1) * pipe.mel.hop_length)
											assert image.height == pipe.unet.config.sample_size[0] and image.width == pipe.unet.config.sample_size[1]
											A__	:							List[Any]						    = np.frombuffer(image.tobytes()			,		dtype="uint8")[:10]
											A__	:							str						    = np.array([151, 167, 154, 144, 122, 134, 121, 105, 70, 26])

											assert np.abs(image_slice.flatten() - expected_slice).max() == 0




from collections.abc import Callable

import numpy as np
def 	snake_case__       (					__lowercase							,     __lowercase							,     __lowercase							,     __lowercase							,     __lowercase       )		->		np.array:







						"""simple docstring"""

						A__	:							Any						    = int(np.ceil((x_end - xa) / step_size       )       )
						A__	:							Union[str, Any]						    = np.zeros((n + 1,)       )
						A__	:							Any						    = ya
						A__	:							Union[str, Any]						    = xa

						for k in range(__lowercase       ):
												A__	:							Any						    = y[k] + step_size * ode_func(__lowercase							,     y[k]       )
												A__	:							Any						    = y[k] + (
												    (step_size / 2) * (ode_func(__lowercase							,     y[k]       ) + ode_func(x + step_size							,     __lowercase       ))
												)
												x += step_size

						return y


if __name__ == "__main__":
		import doctest

		doctest.testmod()