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

						def  A_				(			self				)					->      Optional[Any]:


								'''simple docstring'''

								torch.manual_seed(0				)
								_UpperCamelCase						    =	UNetaDModel(
								    block_out_channels=(32, 32, 64)       ,							extra_in_channels=16       ,							sample_size=5_12       ,							sample_rate=1_60_00       ,							in_channels=2       ,							out_channels=2       ,							flip_sin_to_cos=a       ,							use_timestep_embedding=a       ,							time_embedding_type="""fourier"""       ,							mid_block_type="""UNetMidBlock1D"""       ,							down_block_types=("""DownBlock1DNoSkip""", """DownBlock1D""", """AttnDownBlock1D""")       ,							up_block_types=("""AttnUpBlock1D""", """UpBlock1D""", """UpBlock1DNoSkip""")       ,							)
								_UpperCamelCase						    =	IPNDMScheduler()

								_UpperCamelCase						    =	{
								    '''unet''': unet,
								    '''scheduler''': scheduler,
								}
								return components

						def  A_				(			self       ,							a       ,							a=0				)					->      Any:


								'''simple docstring'''

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

						def  A_				(			self				)					->      Dict:


								'''simple docstring'''

								_UpperCamelCase						    =	'''cpu'''  # ensure determinism for the device-dependent torch.Generator
								_UpperCamelCase						    =	self.get_dummy_components()
								_UpperCamelCase						    =	DanceDiffusionPipeline(**a				)
								_UpperCamelCase						    =	pipe.to(a				)
								pipe.set_progress_bar_config(disable=a				)

								_UpperCamelCase						    =	self.get_dummy_inputs(a				)
								_UpperCamelCase						    =	pipe(**a				)
								_UpperCamelCase						    =	output.audios

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

								assert audio.shape == (1, 2, components["unet"].sample_size)
								_UpperCamelCase						    =	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				)					->      List[Any]:


								'''simple docstring'''

								return super().test_save_load_local()

						@skip_mps
						def  A_				(			self				)					->      Tuple:


								'''simple docstring'''

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

						@skip_mps
						def  A_				(			self				)					->      Any:


								'''simple docstring'''

								return super().test_save_load_optional_components()

						@skip_mps
						def  A_				(			self				)					->      Any:


								'''simple docstring'''

								return super().test_attention_slicing_forward_pass()







						def  A_				(			self				)					->      Dict:


								'''simple docstring'''

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







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

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


								'''simple docstring'''

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

						def  A_				(			self				)					->      List[Any]:


								'''simple docstring'''

								_UpperCamelCase						    =	torch_device

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

								_UpperCamelCase						    =	torch.manual_seed(0				)
								_UpperCamelCase						    =	pipe(generator=a       ,							num_inference_steps=1_00       ,							audio_length_in_s=4.096				)
								_UpperCamelCase						    =	output.audios

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

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


								'''simple docstring'''

								_UpperCamelCase						    =	torch_device

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

								_UpperCamelCase						    =	torch.manual_seed(0				)
								_UpperCamelCase						    =	pipe(generator=a       ,							num_inference_steps=1_00       ,							audio_length_in_s=4.096				)
								_UpperCamelCase						    =	output.audios

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

								assert audio.shape == (1, 2, pipe.unet.sample_size)
								_UpperCamelCase						    =	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 warnings
from typing import List, Optional, Union

from ...processing_utils import ProcessorMixin
from ...tokenization_utils_base import BatchEncoding, PaddingStrategy, PreTokenizedInput, TextInput, TruncationStrategy
from ...utils import TensorType







class        lowerCAmelCase__      (   __lowercase	):
						UpperCamelCase_							: Dict	=		["image_processor", "tokenizer"]
						UpperCamelCase_							: Union[str, Any]	=		"ViltImageProcessor"
						UpperCamelCase_							: List[str]	=		("BertTokenizer", "BertTokenizerFast")

						def __init__(			self       ,							a=None       ,							a=None       ,							**a				)					->      Optional[int]:


								'''simple docstring'''

								_UpperCamelCase						    =	None
								if "feature_extractor" in kwargs:
										warnings.warn(
										    """The `feature_extractor` argument is deprecated and will be removed in v5, use `image_processor`"""
										    """ instead."""       ,							a       ,							)
										_UpperCamelCase						    =	kwargs.pop("""feature_extractor"""				)

								_UpperCamelCase						    =	image_processor if image_processor is not None else feature_extractor
								if image_processor is None:
										raise ValueError("""You need to specify an `image_processor`."""				)
								if tokenizer is None:
										raise ValueError("""You need to specify a `tokenizer`."""				)

								super().__init__(a       ,							a				)
								_UpperCamelCase						    =	self.image_processor

						def __call__(			self       ,							a       ,							a = None       ,							a = True       ,							a = False       ,							a = None       ,							a = None       ,							a = 0       ,							a = None       ,							a = None       ,							a = None       ,							a = False       ,							a = False       ,							a = False       ,							a = False       ,							a = True       ,							a = None       ,							**a       ,							)					->      BatchEncoding:


								'''simple docstring'''

								_UpperCamelCase						    =	self.tokenizer(
								    text=a       ,							add_special_tokens=a       ,							padding=a       ,							truncation=a       ,							max_length=a       ,							stride=a       ,							pad_to_multiple_of=a       ,							return_token_type_ids=a       ,							return_attention_mask=a       ,							return_overflowing_tokens=a       ,							return_special_tokens_mask=a       ,							return_offsets_mapping=a       ,							return_length=a       ,							verbose=a       ,							return_tensors=a       ,							**a       ,							)
								# add pixel_values + pixel_mask
								_UpperCamelCase						    =	self.image_processor(a       ,							return_tensors=a				)
								encoding.update(a				)

								return encoding

						def  A_				(			self       ,							*a       ,							**a				)					->      Optional[Any]:


								'''simple docstring'''

								return self.tokenizer.batch_decode(*a       ,							**a				)

						def  A_				(			self       ,							*a       ,							**a				)					->      int:


								'''simple docstring'''

								return self.tokenizer.decode(*a       ,							**a				)

						@property
						def  A_				(			self				)					->      Optional[Any]:


								'''simple docstring'''

								_UpperCamelCase						    =	self.tokenizer.model_input_names
								_UpperCamelCase						    =	self.image_processor.model_input_names
								return list(dict.fromkeys(tokenizer_input_names + image_processor_input_names				)				)

						@property
						def  A_				(			self				)					->      Tuple:


								'''simple docstring'''

								warnings.warn(
								    """`feature_extractor_class` is deprecated and will be removed in v5. Use `image_processor_class` instead."""       ,							a       ,							)
								return self.image_processor_class







						@property
						def  A_				(			self				)					->      Dict:


								'''simple docstring'''

								warnings.warn(
								    """`feature_extractor` is deprecated and will be removed in v5. Use `image_processor` instead."""       ,							a       ,							)
								return self.image_processor



