
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     _SCREAMING_SNAKE_CASE     (      snake_case_			,       unittest.TestCase						):
	lowerCAmelCase__         = DanceDiffusionPipeline
	lowerCAmelCase__         = UNCONDITIONAL_AUDIO_GENERATION_PARAMS
	lowerCAmelCase__         = PipelineTesterMixin.required_optional_params - {
	    'callback',
	    'latents',
	    'callback_steps',
	    'output_type',
	    'num_images_per_prompt',
	}
	lowerCAmelCase__         = UNCONDITIONAL_AUDIO_GENERATION_BATCH_PARAMS
	lowerCAmelCase__         = False
	lowerCAmelCase__         = False



	def     SCREAMING_SNAKE_CASE_(    self )				->       Optional[Any]:
					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=lowercase						,  use_timestep_embedding=lowercase						,  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     SCREAMING_SNAKE_CASE_(    self						,  lowercase						,  lowercase=0 )				->       Tuple:
					if str(lowercase ).startswith("mps" ):
									lowerCamelCase_						=			torch.manual_seed(lowercase )
					else:
									lowerCamelCase_						=			torch.Generator(device=lowercase ).manual_seed(lowercase )
					lowerCamelCase_						=			{
					    "batch_size": 1,
					    "generator": generator,
					    "num_inference_steps": 4,
					}
					return inputs



	def     SCREAMING_SNAKE_CASE_(    self )				->       str:
					lowerCamelCase_						=			"cpu"  # ensure determinism for the device-dependent torch.Generator
					lowerCamelCase_						=			self.get_dummy_components()
					lowerCamelCase_						=			DanceDiffusionPipeline(**lowercase )
					lowerCamelCase_						=			pipe.to(lowercase )
					pipe.set_progress_bar_config(disable=lowercase )

					lowerCamelCase_						=			self.get_dummy_inputs(lowercase )
					lowerCamelCase_						=			pipe(**lowercase )
					lowerCamelCase_						=			output.audios

					lowerCamelCase_						=			audio[0, -3:, -3:]

					assert audio.shape == (1, 2, components["unet"].sample_size)
					lowerCamelCase_						=			np.array([-0.7_2_6_5, 1.0_0_0_0, -0.8_3_8_8, 0.1_1_7_5, 0.9_4_9_8, -1.0_0_0_0] )
					assert np.abs(audio_slice.flatten() - expected_slice ).max() < 1e-2



	@skip_mps
	def     SCREAMING_SNAKE_CASE_(    self )				->       Dict:
					return super().test_save_load_local()



	@skip_mps
	def     SCREAMING_SNAKE_CASE_(    self )				->       int:
					return super().test_dict_tuple_outputs_equivalent(expected_max_difference=3e-3 )



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



	@skip_mps
	def     SCREAMING_SNAKE_CASE_(    self )				->       Optional[Any]:
					return super().test_attention_slicing_forward_pass()



	def     SCREAMING_SNAKE_CASE_(    self )				->       Optional[Any]:
					super().test_inference_batch_single_identical(expected_max_diff=3e-3 )




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



	def     SCREAMING_SNAKE_CASE_(    self )				->       List[str]:
					# clean up the VRAM after each test
					super().tearDown()
					gc.collect()
					torch.cuda.empty_cache()



	def     SCREAMING_SNAKE_CASE_(    self )				->       Any:
					lowerCamelCase_						=			torch_device

					lowerCamelCase_						=			DanceDiffusionPipeline.from_pretrained("harmonai/maestro-150k" )
					lowerCamelCase_						=			pipe.to(lowercase )
					pipe.set_progress_bar_config(disable=lowercase )

					lowerCamelCase_						=			torch.manual_seed(0 )
					lowerCamelCase_						=			pipe(generator=lowercase						,  num_inference_steps=100						,  audio_length_in_s=4.0_9_6 )
					lowerCamelCase_						=			output.audios

					lowerCamelCase_						=			audio[0, -3:, -3:]

					assert audio.shape == (1, 2, pipe.unet.sample_size)
					lowerCamelCase_						=			np.array([-0.0_1_9_2, -0.0_2_3_1, -0.0_3_1_8, -0.0_0_5_9, 0.0_0_0_2, -0.0_0_2_0] )

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



	def     SCREAMING_SNAKE_CASE_(    self )				->       List[Any]:
					lowerCamelCase_						=			torch_device

					lowerCamelCase_						=			DanceDiffusionPipeline.from_pretrained("harmonai/maestro-150k"						,  torch_dtype=torch.floataa )
					lowerCamelCase_						=			pipe.to(lowercase )
					pipe.set_progress_bar_config(disable=lowercase )

					lowerCamelCase_						=			torch.manual_seed(0 )
					lowerCamelCase_						=			pipe(generator=lowercase						,  num_inference_steps=100						,  audio_length_in_s=4.0_9_6 )
					lowerCamelCase_						=			output.audios

					lowerCamelCase_						=			audio[0, -3:, -3:]

					assert audio.shape == (1, 2, pipe.unet.sample_size)
					lowerCamelCase_						=			np.array([-0.0_3_6_7, -0.0_4_8_8, -0.0_7_7_1, -0.0_5_2_5, -0.0_4_4_4, -0.0_3_4_1] )

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




from collections import deque





def 			lowerCamelCase_ (				lowerCamelCase__    ):
				lowerCamelCase_						=			len(lowerCamelCase__    )
				lowerCamelCase_						=			deque()
				lowerCamelCase_						=			[False for _ in range(lowerCamelCase__    )]
				lowerCamelCase_						=			[-1 for _ in range(lowerCamelCase__    )]
				lowerCamelCase_						=			index_of[:]

				def strong_connect(lowerCamelCase__	, lowerCamelCase__	, lowerCamelCase__    ):
								lowerCamelCase_						=			index  # the number when this node is seen
								lowerCamelCase_						=			index  # lowest rank node reachable from here
								index += 1
								stack.append(lowerCamelCase__    )
								lowerCamelCase_						=			True

								for w in g[v]:
												if index_of[w] == -1:
																lowerCamelCase_						=			strong_connect(lowerCamelCase__	, lowerCamelCase__	, lowerCamelCase__    )
																lowerCamelCase_						=			(
																    lowlink_of[w] if lowlink_of[w] < lowlink_of[v] else lowlink_of[v]
																)
												elif on_stack[w]:
																lowerCamelCase_						=			(
																    lowlink_of[w] if lowlink_of[w] < lowlink_of[v] else lowlink_of[v]
																)

								if lowlink_of[v] == index_of[v]:
												lowerCamelCase_						=			[]
												lowerCamelCase_						=			stack.pop()
												lowerCamelCase_						=			False
												component.append(lowerCamelCase__    )
												while w != v:
																lowerCamelCase_						=			stack.pop()
																lowerCamelCase_						=			False
																component.append(lowerCamelCase__    )
												components.append(lowerCamelCase__    )
								return index

				lowerCamelCase_						=			[]
				for v in range(lowerCamelCase__    ):
								if index_of[v] == -1:
												strong_connect(lowerCamelCase__	, 0	, lowerCamelCase__    )

				return components





def 			lowerCamelCase_ (				lowerCamelCase__	, lowerCamelCase__    ):
				lowerCamelCase_						=			[[] for _ in range(lowerCamelCase__    )]
				for u, v in edges:
								g[u].append(lowerCamelCase__    )
				return g


if __name__ == "__main__":
			# Test
			__A				=7
			__A				=[0, 0, 1, 2, 3, 3, 4, 4, 6]
			__A				=[1, 3, 2, 0, 1, 4, 5, 6, 5]
			__A				=[(u, v) for u, v in zip(source, target)]
			__A				=create_graph(n_vertices, edges)

			assert [[5], [6], [4], [3, 2, 1, 0]] == tarjan(g)


