





"""simple docstring"""



import os
import re
import shutil
import sys
import tempfile
import unittest

import black


a_					=     os.path.abspath(os.path.dirname(os.path.dirname(os.path.dirname(__file__))))
sys.path.append(os.path.join(git_repo_path, """utils"""))

import check_copies  # noqa: E402


# This is the reference code that will be used in the tests.
# If DDPMSchedulerOutput is changed in scheduling_ddpm.py, this code needs to be manually updated.
a_					=     """    \"\"\"
    Output class for the scheduler's step function output.

    Args:
        prev_sample (`torch.FloatTensor` of shape `(batch_size, num_channels, height, width)` for images):
            Computed sample (x_{t-1}) of previous timestep. `prev_sample` should be used as next model input in the
            denoising loop.
        pred_original_sample (`torch.FloatTensor` of shape `(batch_size, num_channels, height, width)` for images):
            The predicted denoised sample (x_{0}) based on the model output from the current timestep.
            `pred_original_sample` can be used to preview progress or for guidance.
    \"\"\"

    prev_sample: torch.FloatTensor
    pred_original_sample: Optional[torch.FloatTensor] = None
"""






class 						__snake_case   (						unittest.TestCase	):





 """simple docstring"""


 def    UpperCamelCase__(       self    ):



     '''simple docstring'''




     __A       :		Any							=	tempfile.mkdtemp()
     os.makedirs(os.path.join(self.diffusers_dir   ,		'''schedulers/'''    )    )
     __A       :		str							=	self.diffusers_dir
     shutil.copy(
         os.path.join(__lowerCamelCase   ,		'''src/diffusers/schedulers/scheduling_ddpm.py'''    )   ,		os.path.join(self.diffusers_dir   ,		'''schedulers/scheduling_ddpm.py'''    )   ,		)

 def    UpperCamelCase__(       self    ):



     '''simple docstring'''




     __A       :		List[Any]							=	'''src/diffusers'''
     shutil.rmtree(self.diffusers_dir    )

 def    UpperCamelCase__(       self   ,		__lowerCamelCase   ,		__lowerCamelCase   ,		__lowerCamelCase   ,		__lowerCamelCase=None    ):



     '''simple docstring'''




     __A       :		List[Any]							=	comment + F"""\nclass {class_name}(nn.Module):\n""" + class_code
     if overwrite_result is not None:
         __A       :		str							=	comment + F"""\nclass {class_name}(nn.Module):\n""" + overwrite_result
     __A       :		str							=	black.Mode(target_versions={black.TargetVersion.PYaa}   ,		line_length=119    )
     __A       :		str							=	black.format_str(__lowerCamelCase   ,		mode=__lowerCamelCase    )
     __A       :		Optional[int]							=	os.path.join(self.diffusers_dir   ,		'''new_code.py'''    )
     with open(__lowerCamelCase   ,		'''w'''   ,		newline='''\n'''    ) as f:
         f.write(__lowerCamelCase    )
     if overwrite_result is None:
         self.assertTrue(len(check_copies.is_copy_consistent(__lowerCamelCase    )    ) == 0    )
     else:
         check_copies.is_copy_consistent(f.name   ,		overwrite=__lowerCamelCase    )
         with open(__lowerCamelCase   ,		'''r'''    ) as f:
             self.assertTrue(f.read()   ,		__lowerCamelCase    )

 def    UpperCamelCase__(       self    ):



     '''simple docstring'''




     __A       :		Tuple							=	check_copies.find_code_in_diffusers('''schedulers.scheduling_ddpm.DDPMSchedulerOutput'''    )
     self.assertEqual(__lowerCamelCase   ,		__lowerCamelCase    )







 def    UpperCamelCase__(       self    ):



     '''simple docstring'''




     self.check_copy_consistency(
         '''# Copied from diffusers.schedulers.scheduling_ddpm.DDPMSchedulerOutput'''   ,		'''DDPMSchedulerOutput'''   ,		REFERENCE_CODE + '''\n'''   ,		)

     # With no empty line at the end
     self.check_copy_consistency(
         '''# Copied from diffusers.schedulers.scheduling_ddpm.DDPMSchedulerOutput'''   ,		'''DDPMSchedulerOutput'''   ,		__lowerCamelCase   ,		)

     # Copy consistency with rename
     self.check_copy_consistency(
         '''# Copied from diffusers.schedulers.scheduling_ddpm.DDPMSchedulerOutput with DDPM->Test'''   ,		'''TestSchedulerOutput'''   ,		re.sub('''DDPM'''   ,		'''Test'''   ,		__lowerCamelCase    )   ,		)

     # Copy consistency with a really long name
     __A       :		List[str]							=	'''TestClassWithAReallyLongNameBecauseSomePeopleLikeThatForSomeReason'''
     self.check_copy_consistency(
         F"""# Copied from diffusers.schedulers.scheduling_ddpm.DDPMSchedulerOutput with DDPM->{long_class_name}"""   ,		F"""{long_class_name}SchedulerOutput"""   ,		re.sub('''Bert'''   ,		__lowerCamelCase   ,		__lowerCamelCase    )   ,		)

     # Copy consistency with overwrite
     self.check_copy_consistency(
         '''# Copied from diffusers.schedulers.scheduling_ddpm.DDPMSchedulerOutput with DDPM->Test'''   ,		'''TestSchedulerOutput'''   ,		__lowerCamelCase   ,		overwrite_result=re.sub('''DDPM'''   ,		'''Test'''   ,		__lowerCamelCase    )   ,		)











"""simple docstring"""



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 						__snake_case   (						unittest.TestCase	):





 """simple docstring"""


 def    UpperCamelCase__(       self    ):



     '''simple docstring'''




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

 @property
 def    UpperCamelCase__(       self    ):



     '''simple docstring'''




     torch.manual_seed(0    )
     __A       :		List[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    UpperCamelCase__(       self    ):



     '''simple docstring'''




     torch.manual_seed(0    )
     __A       :		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    UpperCamelCase__(       self    ):



     '''simple docstring'''




     torch.manual_seed(0    )
     __A       :		int							=	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       :		int							=	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    UpperCamelCase__(       self    ):



     '''simple docstring'''




     __A       :		str							=	'''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       :		List[Any]							=	DDPMScheduler()
     __A       :		Optional[Any]							=	AudioDiffusionPipeline(vqvae=__lowerCamelCase   ,		unet=self.dummy_unet   ,		mel=__lowerCamelCase   ,		scheduler=__lowerCamelCase    )
     __A       :		Tuple							=	pipe.to(__lowerCamelCase    )
     pipe.set_progress_bar_config(disable=__lowerCamelCase    )

     __A       :		Tuple							=	torch.Generator(device=__lowerCamelCase    ).manual_seed(42    )
     __A       :		List[str]							=	pipe(generator=__lowerCamelCase   ,		steps=4    )
     __A       :		Union[str, Any]							=	output.audios[0]
     __A       :		str							=	output.images[0]

     __A       :		List[Any]							=	torch.Generator(device=__lowerCamelCase    ).manual_seed(42    )
     __A       :		Any							=	pipe(generator=__lowerCamelCase   ,		steps=4   ,		return_dict=__lowerCamelCase    )
     __A       :		int							=	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       :		Tuple							=	np.frombuffer(image.tobytes()   ,		dtype='''uint8'''    )[:10]
     __A       :		str							=	np.frombuffer(image_from_tuple.tobytes()   ,		dtype='''uint8'''    )[:10]
     __A       :		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       :		int							=	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       :		Any							=	DDIMScheduler()
     __A       :		Optional[Any]							=	self.dummy_vqvae_and_unet
     __A       :		int							=	AudioDiffusionPipeline(
         vqvae=self.dummy_vqvae_and_unet[0]   ,		unet=dummy_vqvae_and_unet[1]   ,		mel=__lowerCamelCase   ,		scheduler=__lowerCamelCase    )
     __A       :		Any							=	pipe.to(__lowerCamelCase    )
     pipe.set_progress_bar_config(disable=__lowerCamelCase    )

     np.random.seed(0    )
     __A       :		str							=	np.random.uniform(-1   ,		1   ,		((dummy_vqvae_and_unet[0].config.sample_size[1] - 1) * mel.hop_length,)    )
     __A       :		Union[str, Any]							=	torch.Generator(device=__lowerCamelCase    ).manual_seed(42    )
     __A       :		int							=	pipe(raw_audio=__lowerCamelCase   ,		generator=__lowerCamelCase   ,		start_step=5   ,		steps=10    )
     __A       :		Any							=	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       :		str							=	np.frombuffer(image.tobytes()   ,		dtype='''uint8'''    )[:10]
     __A       :		Optional[Any]							=	np.array([120, 117, 110, 109, 138, 167, 138, 148, 132, 121]    )

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

     __A       :		Union[str, Any]							=	self.dummy_unet_condition
     __A       :		List[str]							=	AudioDiffusionPipeline(
         vqvae=self.dummy_vqvae_and_unet[0]   ,		unet=__lowerCamelCase   ,		mel=__lowerCamelCase   ,		scheduler=__lowerCamelCase    )
     __A       :		Dict							=	pipe.to(__lowerCamelCase    )
     pipe.set_progress_bar_config(disable=__lowerCamelCase    )

     np.random.seed(0    )
     __A       :		int							=	torch.rand((1, 1, 10)    )
     __A       :		Optional[Any]							=	pipe(generator=__lowerCamelCase   ,		encoding=__lowerCamelCase    )
     __A       :		str							=	output.images[0]
     __A       :		Optional[Any]							=	np.frombuffer(image.tobytes()   ,		dtype='''uint8'''    )[:10]
     __A       :		Tuple							=	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 						__snake_case   (						unittest.TestCase	):





 """simple docstring"""


 def    UpperCamelCase__(       self    ):



     '''simple docstring'''




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







 def    UpperCamelCase__(       self    ):



     '''simple docstring'''




     __A       :		int							=	torch_device

     __A       :		List[Any]							=	DiffusionPipeline.from_pretrained('''teticio/audio-diffusion-ddim-256'''    )
     __A       :		List[Any]							=	pipe.to(__lowerCamelCase    )
     pipe.set_progress_bar_config(disable=__lowerCamelCase    )

     __A       :		Dict							=	torch.Generator(device=__lowerCamelCase    ).manual_seed(42    )
     __A       :		str							=	pipe(generator=__lowerCamelCase    )
     __A       :		Dict							=	output.audios[0]
     __A       :		List[Any]							=	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       :		Optional[int]							=	np.frombuffer(image.tobytes()   ,		dtype='''uint8'''    )[:10]
     __A       :		Optional[Any]							=	np.array([151, 167, 154, 144, 122, 134, 121, 105, 70, 26]    )

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




