
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					)							->			List[str]:
					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					)							->			Any:
					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					)							->			List[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					)							->			int:
					return super().test_save_load_local()



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



	@skip_mps
	def        SCREAMING_SNAKE_CASE_(       self					)							->			Tuple:
					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					)							->			List[str]:
					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					)							->			str:
					# clean up the VRAM after each test
					super().tearDown()
					gc.collect()
					torch.cuda.empty_cache()



	def        SCREAMING_SNAKE_CASE_(       self					)							->			Dict:
					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[str]:
					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




import numpy as np
import torch
import torch.nn as nn
from transformers import CLIPConfig, CLIPVisionModelWithProjection, PreTrainedModel

from ...utils import logging


__A							=logging.get_logger(__name__)




class        _SCREAMING_SNAKE_CASE	( snake_case_		):
	lowerCAmelCase__   =					CLIPConfig

	lowerCAmelCase__   =					['CLIPEncoderLayer']



	def __init__(       self		,    lowercase					)							->			int:
					super().__init__(lowercase					)

					lowerCamelCase_         =     CLIPVisionModelWithProjection(config.vision_config					)

					lowerCamelCase_         =     nn.Linear(config.vision_config.projection_dim		,    1					)
					lowerCamelCase_         =     nn.Linear(config.vision_config.projection_dim		,    1					)



	@torch.no_grad()
	def        SCREAMING_SNAKE_CASE_(       self		,    lowercase		,    lowercase		,    lowercase=0.5		,    lowercase=0.5					)							->			int:
					lowerCamelCase_         =     self.vision_model(lowercase					)[0]

					lowerCamelCase_         =     self.p_head(lowercase					)
					lowerCamelCase_         =     nsfw_detected.flatten()
					lowerCamelCase_         =     nsfw_detected > p_threshold
					lowerCamelCase_         =     nsfw_detected.tolist()

					if any(lowercase					):
									logger.warning(
									    "Potential NSFW content was detected in one or more images. A black image will be returned instead."
									    " Try again with a different prompt and/or seed."					)

					for idx, nsfw_detected_ in enumerate(lowercase					):
									if nsfw_detected_:
													lowerCamelCase_         =     np.zeros(images[idx].shape					)

					lowerCamelCase_         =     self.w_head(lowercase					)
					lowerCamelCase_         =     watermark_detected.flatten()
					lowerCamelCase_         =     watermark_detected > w_threshold
					lowerCamelCase_         =     watermark_detected.tolist()

					if any(lowercase					):
									logger.warning(
									    "Potential watermarked content was detected in one or more images. A black image will be returned instead."
									    " Try again with a different prompt and/or seed."					)

					for idx, watermark_detected_ in enumerate(lowercase					):
									if watermark_detected_:
													lowerCamelCase_         =     np.zeros(images[idx].shape					)

					return images, nsfw_detected, watermark_detected


