



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 				_snake_case			(					lowercase__    ,	unittest.TestCase):
				UpperCamelCase__				:					Tuple     		=DanceDiffusionPipeline
				UpperCamelCase__				:					Any     		=UNCONDITIONAL_AUDIO_GENERATION_PARAMS
				UpperCamelCase__				:					Optional[Any]     		=PipelineTesterMixin.required_optional_params - {
				    """callback""",
				    """latents""",
				    """callback_steps""",
				    """output_type""",
				    """num_images_per_prompt""",
				}
				UpperCamelCase__				:					str     		=UNCONDITIONAL_AUDIO_GENERATION_BATCH_PARAMS
				UpperCamelCase__				:					Optional[Any]     		=False
				UpperCamelCase__				:					List[Any]     		=False




				def 							A__      (						self					:						int						):
							torch.manual_seed(0						)
							lowercase__					  =				UNetaDModel(
							    block_out_channels=(32, 32, 64),					extra_in_channels=16,					sample_size=512,					sample_rate=1_6000,					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"),					)
							lowercase__					  =				IPNDMScheduler()

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




				def 							A__      (						self					:						int,					__lowercase					:						int,					__lowercase					:						Optional[Any]=0						):
							if str(__lowercase						).startswith("mps"						):
										lowercase__					  =				torch.manual_seed(__lowercase						)
							else:
										lowercase__					  =				torch.Generator(device=__lowercase						).manual_seed(__lowercase						)
							lowercase__					  =				{
							    "batch_size": 1,
							    "generator": generator,
							    "num_inference_steps": 4,
							}
							return inputs




				def 							A__      (						self					:						List[str]						):
							lowercase__					  =				"cpu"  # ensure determinism for the device-dependent torch.Generator
							lowercase__					  =				self.get_dummy_components()
							lowercase__					  =				DanceDiffusionPipeline(**__lowercase						)
							lowercase__					  =				pipe.to(__lowercase						)
							pipe.set_progress_bar_config(disable=__lowercase						)

							lowercase__					  =				self.get_dummy_inputs(__lowercase						)
							lowercase__					  =				pipe(**__lowercase						)
							lowercase__					  =				output.audios

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

							assert audio.shape == (1, 2, components["unet"].sample_size)
							lowercase__					  =				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 							A__      (						self					:						Union[str, Any]						):
							return super().test_save_load_local()




				@skip_mps
				def 							A__      (						self					:						Union[str, Any]						):
							return super().test_dict_tuple_outputs_equivalent(expected_max_difference=3e-3						)




				@skip_mps
				def 							A__      (						self					:						Dict						):
							return super().test_save_load_optional_components()




				@skip_mps
				def 							A__      (						self					:						Dict						):
							return super().test_attention_slicing_forward_pass()




				def 							A__      (						self					:						Union[str, Any]						):
							super().test_inference_batch_single_identical(expected_max_diff=3e-3						)
@slow
@require_torch_gpu
class 				_snake_case			(					unittest.TestCase):




				def 							A__      (						self					:						Union[str, Any]						):
							# clean up the VRAM after each test
							super().tearDown()
							gc.collect()
							torch.cuda.empty_cache()




				def 							A__      (						self					:						Any						):
							lowercase__					  =				torch_device

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

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

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

							assert audio.shape == (1, 2, pipe.unet.sample_size)
							lowercase__					  =				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 							A__      (						self					:						str						):
							lowercase__					  =				torch_device

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

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

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

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









import json
from typing import Iterator, List, Union

from tokenizers import AddedToken, Regex, Tokenizer, decoders, normalizers, pre_tokenizers, trainers
from tokenizers.implementations.base_tokenizer import BaseTokenizer
from tokenizers.models import Unigram
from tokenizers.processors import TemplateProcessing
class 				_snake_case			(					lowercase__):




				def __init__(						self					:						Optional[Any],					__lowercase					:						str = "▁",					__lowercase					:						bool = True,					__lowercase					:						Union[str, AddedToken] = "<unk>",					__lowercase					:						Union[str, AddedToken] = "</s>",					__lowercase					:						Union[str, AddedToken] = "<pad>",					):
							lowercase__					  =				{
							    "pad": {"id": 0, "token": pad_token},
							    "eos": {"id": 1, "token": eos_token},
							    "unk": {"id": 2, "token": unk_token},
							}

							lowercase__					  =				[None] * len(self.special_tokens						)
							for token_dict in self.special_tokens.values():
										lowercase__					  =				token_dict["token"]

							lowercase__					  =				Tokenizer(Unigram()						)

							lowercase__					  =				normalizers.Sequence(
							    [
							        normalizers.Nmt(),
							        normalizers.NFKC(),
							        normalizers.Replace(Regex(" {2,}"						),					" "						),
							        normalizers.Lowercase(),
							    ]						)
							lowercase__					  =				pre_tokenizers.Sequence(
							    [
							        pre_tokenizers.Metaspace(replacement=__lowercase,					add_prefix_space=__lowercase						),
							        pre_tokenizers.Digits(individual_digits=__lowercase						),
							        pre_tokenizers.Punctuation(),
							    ]						)
							lowercase__					  =				decoders.Metaspace(replacement=__lowercase,					add_prefix_space=__lowercase						)

							lowercase__					  =				TemplateProcessing(
							    single=F'''$A {self.special_tokens["eos"]["token"]}''',					special_tokens=[(self.special_tokens["eos"]["token"], self.special_tokens["eos"]["id"])],					)

							lowercase__					  =				{
							    "model": "SentencePieceUnigram",
							    "replacement": replacement,
							    "add_prefix_space": add_prefix_space,
							}

							super().__init__(__lowercase,					__lowercase						)




				def 							A__      (						self					:						Union[str, Any],					__lowercase					:						Union[str, List[str]],					__lowercase					:						int = 8000,					__lowercase					:						bool = True,					):
							lowercase__					  =				trainers.UnigramTrainer(
							    vocab_size=__lowercase,					special_tokens=self.special_tokens_list,					show_progress=__lowercase,					)

							if isinstance(__lowercase,					__lowercase						):
										lowercase__					  =				[files]
							self._tokenizer.train(__lowercase,					trainer=__lowercase						)

							self.add_unk_id()




				def 							A__      (						self					:						List[Any],					__lowercase					:						Union[Iterator[str], Iterator[Iterator[str]]],					__lowercase					:						int = 8000,					__lowercase					:						bool = True,					):
							lowercase__					  =				trainers.UnigramTrainer(
							    vocab_size=__lowercase,					special_tokens=self.special_tokens_list,					show_progress=__lowercase,					)

							self._tokenizer.train_from_iterator(__lowercase,					trainer=__lowercase						)

							self.add_unk_id()




				def 							A__      (						self					:						str						):
							lowercase__					  =				json.loads(self._tokenizer.to_str()						)

							lowercase__					  =				self.special_tokens["unk"]["id"]

							lowercase__					  =				Tokenizer.from_str(json.dumps(__lowercase						)						)




