



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 				__A					( A_					, unittest.TestCase      ):
			UpperCamelCase					:Optional[Any]     =			DanceDiffusionPipeline
			UpperCamelCase					:Dict     =			UNCONDITIONAL_AUDIO_GENERATION_PARAMS
			UpperCamelCase					:Dict     =			PipelineTesterMixin.required_optional_params - {
			    '''callback''',
			    '''latents''',
			    '''callback_steps''',
			    '''output_type''',
			    '''num_images_per_prompt''',
			}
			UpperCamelCase					:List[Any]     =			UNCONDITIONAL_AUDIO_GENERATION_BATCH_PARAMS
			UpperCamelCase					:List[Any]     =			False
			UpperCamelCase					:Optional[int]     =			False





			def        _snake_case    (self				):
						torch.manual_seed(0				)
						lowerCamelCase__ :      Tuple								=		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=__magic_name__							,    use_timestep_embedding=__magic_name__							,    time_embedding_type="""fourier"""							,    mid_block_type="""UNetMidBlock1D"""							,    down_block_types=("""DownBlock1DNoSkip""", """DownBlock1D""", """AttnDownBlock1D""")							,    up_block_types=("""AttnUpBlock1D""", """UpBlock1D""", """UpBlock1DNoSkip""")							,    )
						lowerCamelCase__ :      Optional[int]								=		IPNDMScheduler()

						lowerCamelCase__ :      int								=		{
						    """unet""": unet,
						    """scheduler""": scheduler,
						}
						return components





			def        _snake_case    (self							,    __magic_name__							,    __magic_name__=0				):
						if str(__magic_name__				).startswith("""mps"""				):
									lowerCamelCase__ :      Dict								=		torch.manual_seed(__magic_name__				)
						else:
									lowerCamelCase__ :      Dict								=		torch.Generator(device=__magic_name__				).manual_seed(__magic_name__				)
						lowerCamelCase__ :      int								=		{
						    """batch_size""": 1,
						    """generator""": generator,
						    """num_inference_steps""": 4,
						}
						return inputs





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

						lowerCamelCase__ :      Tuple								=		self.get_dummy_inputs(__magic_name__				)
						lowerCamelCase__ :      Dict								=		pipe(**__magic_name__				)
						lowerCamelCase__ :      Optional[int]								=		output.audios

						lowerCamelCase__ :      int								=		audio[0, -3:, -3:]

						assert audio.shape == (1, 2, components["unet"].sample_size)
						lowerCamelCase__ :      Tuple								=		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        _snake_case    (self				):
						return super().test_save_load_local()





			@skip_mps
			def        _snake_case    (self				):
						return super().test_dict_tuple_outputs_equivalent(expected_max_difference=3E-3				)





			@skip_mps
			def        _snake_case    (self				):
						return super().test_save_load_optional_components()





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





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







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





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





			def        _snake_case    (self				):
						lowerCamelCase__ :      List[str]								=		torch_device

						lowerCamelCase__ :      int								=		DanceDiffusionPipeline.from_pretrained("""harmonai/maestro-150k"""				)
						lowerCamelCase__ :      Optional[Any]								=		pipe.to(__magic_name__				)
						pipe.set_progress_bar_config(disable=__magic_name__				)

						lowerCamelCase__ :      Optional[int]								=		torch.manual_seed(0				)
						lowerCamelCase__ :      List[Any]								=		pipe(generator=__magic_name__							,    num_inference_steps=100							,    audio_length_in_s=4.0_96				)
						lowerCamelCase__ :      Dict								=		output.audios

						lowerCamelCase__ :      Any								=		audio[0, -3:, -3:]

						assert audio.shape == (1, 2, pipe.unet.sample_size)
						lowerCamelCase__ :      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        _snake_case    (self				):
						lowerCamelCase__ :      Optional[int]								=		torch_device

						lowerCamelCase__ :      Tuple								=		DanceDiffusionPipeline.from_pretrained("""harmonai/maestro-150k"""							,    torch_dtype=torch.floataa				)
						lowerCamelCase__ :      List[Any]								=		pipe.to(__magic_name__				)
						pipe.set_progress_bar_config(disable=__magic_name__				)

						lowerCamelCase__ :      Dict								=		torch.manual_seed(0				)
						lowerCamelCase__ :      Any								=		pipe(generator=__magic_name__							,    num_inference_steps=100							,    audio_length_in_s=4.0_96				)
						lowerCamelCase__ :      str								=		output.audios

						lowerCamelCase__ :      Union[str, Any]								=		audio[0, -3:, -3:]

						assert audio.shape == (1, 2, pipe.unet.sample_size)
						lowerCamelCase__ :      int								=		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








# Author: OMKAR PATHAK, Nwachukwu Chidiebere

# Use a Python dictionary to construct the graph.
from __future__ import annotations

from pprint import pformat
from typing import Generic, TypeVar

_lowercase     							=						TypeVar('''T''')
class 				__A					( Generic[T]      ):





			def __init__(self							,    __magic_name__ = True				):
						lowerCamelCase__ :      dict[T, list[T]]								=		{}  # dictionary of lists
						lowerCamelCase__ :      List[str]								=		directed





			def        _snake_case    (self							,    __magic_name__							,    __magic_name__				):

						if not self.directed:  # For undirected graphs
									# if both source vertex and destination vertex are both present in the
									# adjacency list, add destination vertex to source vertex list of adjacent
									# vertices and add source vertex to destination vertex list of adjacent
									# vertices.
									if source_vertex in self.adj_list and destination_vertex in self.adj_list:
												self.adj_list[source_vertex].append(__magic_name__				)
												self.adj_list[destination_vertex].append(__magic_name__				)
									# if only source vertex is present in adjacency list, add destination vertex
									# to source vertex list of adjacent vertices, then create a new vertex with
									# destination vertex as key and assign a list containing the source vertex
									# as it's first adjacent vertex.
									elif source_vertex in self.adj_list:
												self.adj_list[source_vertex].append(__magic_name__				)
												lowerCamelCase__ :      int								=		[source_vertex]
									# if only destination vertex is present in adjacency list, add source vertex
									# to destination vertex list of adjacent vertices, then create a new vertex
									# with source vertex as key and assign a list containing the source vertex
									# as it's first adjacent vertex.
									elif destination_vertex in self.adj_list:
												self.adj_list[destination_vertex].append(__magic_name__				)
												lowerCamelCase__ :      Dict								=		[destination_vertex]
									# if both source vertex and destination vertex are not present in adjacency
									# list, create a new vertex with source vertex as key and assign a list
									# containing the destination vertex as it's first adjacent vertex also
									# create a new vertex with destination vertex as key and assign a list
									# containing the source vertex as it's first adjacent vertex.
									else:
												lowerCamelCase__ :      Any								=		[destination_vertex]
												lowerCamelCase__ :      Optional[Any]								=		[source_vertex]
						else:  # For directed graphs
									# if both source vertex and destination vertex are present in adjacency
									# list, add destination vertex to source vertex list of adjacent vertices.
									if source_vertex in self.adj_list and destination_vertex in self.adj_list:
												self.adj_list[source_vertex].append(__magic_name__				)
									# if only source vertex is present in adjacency list, add destination
									# vertex to source vertex list of adjacent vertices and create a new vertex
									# with destination vertex as key, which has no adjacent vertex
									elif source_vertex in self.adj_list:
												self.adj_list[source_vertex].append(__magic_name__				)
												lowerCamelCase__ :      Optional[Any]								=		[]
									# if only destination vertex is present in adjacency list, create a new
									# vertex with source vertex as key and assign a list containing destination
									# vertex as first adjacent vertex
									elif destination_vertex in self.adj_list:
												lowerCamelCase__ :      Tuple								=		[destination_vertex]
									# if both source vertex and destination vertex are not present in adjacency
									# list, create a new vertex with source vertex as key and a list containing
									# destination vertex as it's first adjacent vertex. Then create a new vertex
									# with destination vertex as key, which has no adjacent vertex
									else:
												lowerCamelCase__ :      Optional[int]								=		[destination_vertex]
												lowerCamelCase__ :      List[Any]								=		[]

						return self





			def __repr__(self				):
						return pformat(self.adj_list				)



