

from __future__ import annotations

from random import random
from typing import Generic, TypeVar

lowerCAmelCase__	   = TypeVar("""KT""")
lowerCAmelCase__	   = TypeVar("""VT""")

class 					a__						(					Generic[KT, VT]			):


			def __init__(    self      ,      lowercase = "root"      ,      lowercase = None       )  ->						Union[str, Any]:
							'''simple docstring'''






							A__        =			key
							A__        =			value
							A__        =			[]


			def __repr__(    self       )  ->						List[str]:
							'''simple docstring'''






							return F'Node({self.key}: {self.value})'


			@property
			def      UpperCamelCase		(    self       )  ->						Tuple:
							'''simple docstring'''






							return len(self.forward       )

class 					a__						(					Generic[KT, VT]			):


			def __init__(    self      ,      lowercase = 0.5      ,      lowercase = 16       )  ->						Optional[Any]:
							'''simple docstring'''






							A__        =			Node[KT, VT]()
							A__        =			0
							A__        =			p
							A__        =			max_level


			def __str__(    self       )  ->						Dict:
							'''simple docstring'''






							A__        =			list(self       )

							if len(A_       ) == 0:
											return F'SkipList(level={self.level})'

							A__        =			max((len(str(A_       )       ) for item in items)      ,      default=4       )
							A__        =			max(A_      ,      4       ) + 4

							A__        =			self.head
							A__        =			[]

							A__        =			node.forward.copy()
							lines.append(F'[{node.key}]'.ljust(A_      ,      "-"       ) + "* " * len(A_       )       )
							lines.append(" " * label_size + "| " * len(A_       )       )

							while len(node.forward       ) != 0:
											A__        =			node.forward[0]

											lines.append(
											    F'[{node.key}]'.ljust(A_      ,      "-"       )
											    + " ".join(str(n.key       ) if n.key == node.key else "|" for n in forwards       )       )
											lines.append(" " * label_size + "| " * len(A_       )       )
											A__        =			node.forward

							lines.append("None".ljust(A_       ) + "* " * len(A_       )       )
							return F'SkipList(level={self.level})\n' + "\n".join(A_       )


			def __iter__(    self       )  ->						str:
							'''simple docstring'''






							A__        =			self.head

							while len(node.forward       ) != 0:
											yield node.forward[0].key
											A__        =			node.forward[0]


			def      UpperCamelCase		(    self       )  ->						Union[str, Any]:
							'''simple docstring'''






							A__        =			1
							while random() < self.p and level < self.max_level:
											level += 1

							return level


			def      UpperCamelCase		(    self      ,      lowercase       )  ->						List[str]:
							'''simple docstring'''






							A__        =			[]

							A__        =			self.head

							for i in reversed(range(self.level       )       ):
											# i < node.level - When node level is lesser than `i` decrement `i`.
											# node.forward[i].key < key - Jumping to node with key value higher
											#                             or equal to searched key would result
											#                             in skipping searched key.
											while i < node.level and node.forward[i].key < key:
															A__        =			node.forward[i]
											# Each leftmost node (relative to searched node) will potentially have to
											# be updated.
											update_vector.append(A_       )

							update_vector.reverse()  # Note that we were inserting values in reverse order.

							# len(node.forward) != 0 - If current node doesn't contain any further
							#                          references then searched key is not present.
							# node.forward[0].key == key - Next node key should be equal to search key
							#                              if key is present.
							if len(node.forward       ) != 0 and node.forward[0].key == key:
											return node.forward[0], update_vector
							else:
											return None, update_vector


			def      UpperCamelCase		(    self      ,      lowercase       )  ->						Optional[Any]:
							'''simple docstring'''






							A__        =			self._locate_node(A_       )

							if node is not None:
											for i, update_node in enumerate(A_       ):
															# Remove or replace all references to removed node.
															if update_node.level > i and update_node.forward[i].key == key:
																			if node.level > i:
																							A__        =			node.forward[i]
																			else:
																							A__        =			update_node.forward[:i]


			def      UpperCamelCase		(    self      ,      lowercase      ,      lowercase       )  ->						Any:
							'''simple docstring'''






							A__        =			self._locate_node(A_       )
							if node is not None:
											A__        =			value
							else:
											A__        =			self.random_level()

											if level > self.level:
															# After level increase we have to add additional nodes to head.
															for _ in range(self.level - 1      ,      A_       ):
																			update_vector.append(self.head       )
															A__        =			level

											A__        =			Node(A_      ,      A_       )

											for i, update_node in enumerate(update_vector[:level]       ):
															# Change references to pass through new node.
															if update_node.level > i:
																			new_node.forward.append(update_node.forward[i]       )

															if update_node.level < i + 1:
																			update_node.forward.append(A_       )
															else:
																			A__        =			new_node


			def      UpperCamelCase		(    self      ,      lowercase       )  ->						Dict:
							'''simple docstring'''






							A__        =			self._locate_node(A_       )

							if node is not None:
											return node.value

							return None



def 					lowerCAmelCase__							(	)			->      List[Any]:




				'''simple docstring'''





				A__        =			SkipList()
				skip_list.insert("Key1"	,       3      )
				skip_list.insert("Key2"	,       1_2      )
				skip_list.insert("Key3"	,       4_1      )
				skip_list.insert("Key4"	,       -1_9      )

				A__        =			skip_list.head
				A__        =			{}
				while node.level != 0:
								A__        =			node.forward[0]
								A__        =			node.value

				assert len(_lowerCAmelCase      ) == 4
				assert all_values["Key1"] == 3
				assert all_values["Key2"] == 1_2
				assert all_values["Key3"] == 4_1
				assert all_values["Key4"] == -1_9



def 					lowerCAmelCase__							(	)			->      List[Any]:




				'''simple docstring'''





				A__        =			SkipList()
				skip_list.insert("Key1"	,       1_0      )
				skip_list.insert("Key1"	,       1_2      )

				skip_list.insert("Key5"	,       7      )
				skip_list.insert("Key7"	,       1_0      )
				skip_list.insert("Key10"	,       5      )

				skip_list.insert("Key7"	,       7      )
				skip_list.insert("Key5"	,       5      )
				skip_list.insert("Key10"	,       1_0      )

				A__        =			skip_list.head
				A__        =			{}
				while node.level != 0:
								A__        =			node.forward[0]
								A__        =			node.value

				if len(_lowerCAmelCase      ) != 4:
								print()
				assert len(_lowerCAmelCase      ) == 4
				assert all_values["Key1"] == 1_2
				assert all_values["Key7"] == 7
				assert all_values["Key5"] == 5
				assert all_values["Key10"] == 1_0



def 					lowerCAmelCase__							(	)			->      List[Any]:




				'''simple docstring'''





				A__        =			SkipList()
				assert skip_list.find("Some key"      ) is None



def 					lowerCAmelCase__							(	)			->      Tuple:




				'''simple docstring'''





				A__        =			SkipList()

				skip_list.insert("Key2"	,       2_0      )
				assert skip_list.find("Key2"      ) == 2_0

				skip_list.insert("Some Key"	,       1_0      )
				skip_list.insert("Key2"	,       8      )
				skip_list.insert("V"	,       1_3      )

				assert skip_list.find("Y"      ) is None
				assert skip_list.find("Key2"      ) == 8
				assert skip_list.find("Some Key"      ) == 1_0
				assert skip_list.find("V"      ) == 1_3



def 					lowerCAmelCase__							(	)			->      Dict:




				'''simple docstring'''





				A__        =			SkipList()
				skip_list.delete("Some key"      )

				assert len(skip_list.head.forward      ) == 0



def 					lowerCAmelCase__							(	)			->      Dict:




				'''simple docstring'''





				A__        =			SkipList()

				skip_list.insert("Key1"	,       1_2      )
				skip_list.insert("V"	,       1_3      )
				skip_list.insert("X"	,       1_4      )
				skip_list.insert("Key2"	,       1_5      )

				skip_list.delete("V"      )
				skip_list.delete("Key2"      )

				assert skip_list.find("V"      ) is None
				assert skip_list.find("Key2"      ) is None



def 					lowerCAmelCase__							(	)			->      List[str]:




				'''simple docstring'''





				A__        =			SkipList()

				skip_list.insert("Key1"	,       1_2      )
				skip_list.insert("V"	,       1_3      )
				skip_list.insert("X"	,       1_4      )
				skip_list.insert("Key2"	,       1_5      )

				skip_list.delete("V"      )
				assert skip_list.find("V"      ) is None
				assert skip_list.find("X"      ) == 1_4
				assert skip_list.find("Key1"      ) == 1_2
				assert skip_list.find("Key2"      ) == 1_5

				skip_list.delete("X"      )
				assert skip_list.find("V"      ) is None
				assert skip_list.find("X"      ) is None
				assert skip_list.find("Key1"      ) == 1_2
				assert skip_list.find("Key2"      ) == 1_5

				skip_list.delete("Key1"      )
				assert skip_list.find("V"      ) is None
				assert skip_list.find("X"      ) is None
				assert skip_list.find("Key1"      ) is None
				assert skip_list.find("Key2"      ) == 1_5

				skip_list.delete("Key2"      )
				assert skip_list.find("V"      ) is None
				assert skip_list.find("X"      ) is None
				assert skip_list.find("Key1"      ) is None
				assert skip_list.find("Key2"      ) is None



def 					lowerCAmelCase__							(	)			->      List[Any]:




				'''simple docstring'''





				A__        =			SkipList()

				skip_list.insert("Key1"	,       1_2      )
				skip_list.insert("V"	,       1_3      )
				skip_list.insert("X"	,       1_4_2      )
				skip_list.insert("Key2"	,       1_5      )

				skip_list.delete("X"      )

				def traverse_keys(SCREAMING_SNAKE_CASE_:							Tuple      ):
								yield node.key
								for forward_node in node.forward:
												yield from traverse_keys(_lowerCAmelCase      )

				assert len(set(traverse_keys(skip_list.head      )      )      ) == 4



def 					lowerCAmelCase__							(	)			->      Union[str, Any]:




				'''simple docstring'''





				def is_sorted(SCREAMING_SNAKE_CASE_:							str      ):
								return all(next_item >= item for item, next_item in zip(_lowerCAmelCase	,       lst[1:]      )      )

				A__        =			SkipList()
				for i in range(1_0      ):
								skip_list.insert(_lowerCAmelCase	,       _lowerCAmelCase      )
				assert is_sorted(list(_lowerCAmelCase      )      )
				skip_list.delete(5      )
				skip_list.delete(8      )
				skip_list.delete(2      )
				assert is_sorted(list(_lowerCAmelCase      )      )
				skip_list.insert(-1_2	,       -1_2      )
				skip_list.insert(7_7	,       7_7      )
				assert is_sorted(list(_lowerCAmelCase      )      )



def 					lowerCAmelCase__							(	)			->      Tuple:




				'''simple docstring'''





				for _ in range(1_0_0      ):
								# Repeat test 100 times due to the probabilistic nature of skip list
								# random values == random bugs
								test_insert()
								test_insert_overrides_existing_value()

								test_searching_empty_list_returns_none()
								test_search()

								test_deleting_item_from_empty_list_do_nothing()
								test_deleted_items_are_not_founded_by_find_method()
								test_delete_removes_only_given_key()
								test_delete_doesnt_leave_dead_nodes()

								test_iter_always_yields_sorted_values()



def 					lowerCAmelCase__							(	)			->      List[str]:




				'''simple docstring'''





				A__        =			SkipList()
				skip_list.insert(2	,       "2"      )
				skip_list.insert(4	,       "4"      )
				skip_list.insert(6	,       "4"      )
				skip_list.insert(4	,       "5"      )
				skip_list.insert(8	,       "4"      )
				skip_list.insert(9	,       "4"      )

				skip_list.delete(4      )

				print(_lowerCAmelCase      )


if __name__ == "__main__":
	import doctest

	doctest.testmod()
	main()




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__						(					snake_case							,	unittest.TestCase			):
			"""simple docstring"""







			__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      UpperCamelCase		(    self       )  ->						Optional[int]:
							'''simple docstring'''






							torch.manual_seed(0       )
							A__        =			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")      ,      )
							A__        =			IPNDMScheduler()

							A__        =			{
							    "unet": unet,
							    "scheduler": scheduler,
							}
							return components


			def      UpperCamelCase		(    self      ,      lowercase      ,      lowercase=0       )  ->						Union[str, Any]:
							'''simple docstring'''






							if str(lowercase       ).startswith("mps"       ):
											A__        =			torch.manual_seed(lowercase       )
							else:
											A__        =			torch.Generator(device=lowercase       ).manual_seed(lowercase       )
							A__        =			{
							    "batch_size": 1,
							    "generator": generator,
							    "num_inference_steps": 4,
							}
							return inputs


			def      UpperCamelCase		(    self       )  ->						Any:
							'''simple docstring'''






							A__        =			"cpu"  # ensure determinism for the device-dependent torch.Generator
							A__        =			self.get_dummy_components()
							A__        =			DanceDiffusionPipeline(**lowercase       )
							A__        =			pipe.to(lowercase       )
							pipe.set_progress_bar_config(disable=lowercase       )

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

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

							assert audio.shape == (1, 2, components["unet"].sample_size)
							A__        =			np.array([-0.7265, 1.0000, -0.8388, 0.1175, 0.9498, -1.0000]       )
							assert np.abs(audio_slice.flatten() - expected_slice       ).max() < 1e-2


			@skip_mps
			def      UpperCamelCase		(    self       )  ->						Dict:
							'''simple docstring'''






							return super().test_save_load_local()


			@skip_mps
			def      UpperCamelCase		(    self       )  ->						int:
							'''simple docstring'''






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


			@skip_mps
			def      UpperCamelCase		(    self       )  ->						Optional[Any]:
							'''simple docstring'''






							return super().test_save_load_optional_components()


			@skip_mps
			def      UpperCamelCase		(    self       )  ->						int:
							'''simple docstring'''






							return super().test_attention_slicing_forward_pass()


			def      UpperCamelCase		(    self       )  ->						str:
							'''simple docstring'''






							super().test_inference_batch_single_identical(expected_max_diff=3e-3       )

@slow
@require_torch_gpu
class 					a__						(					unittest.TestCase			):
			"""simple docstring"""







			def      UpperCamelCase		(    self       )  ->						int:
							'''simple docstring'''






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


			def      UpperCamelCase		(    self       )  ->						int:
							'''simple docstring'''






							A__        =			torch_device

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

							A__        =			torch.manual_seed(0       )
							A__        =			pipe(generator=lowercase      ,      num_inference_steps=100      ,      audio_length_in_s=4.096       )
							A__        =			output.audios

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

							assert audio.shape == (1, 2, pipe.unet.sample_size)
							A__        =			np.array([-0.0192, -0.0231, -0.0318, -0.0059, 0.0002, -0.0020]       )

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


			def      UpperCamelCase		(    self       )  ->						List[Any]:
							'''simple docstring'''






							A__        =			torch_device

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

							A__        =			torch.manual_seed(0       )
							A__        =			pipe(generator=lowercase      ,      num_inference_steps=100      ,      audio_length_in_s=4.096       )
							A__        =			output.audios

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

							assert audio.shape == (1, 2, pipe.unet.sample_size)
							A__        =			np.array([-0.0367, -0.0488, -0.0771, -0.0525, -0.0444, -0.0341]       )

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

