






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






			def    SCREAMING_SNAKE_CASE__    (       self)      ->       Union[str, Any]:
						torch.manual_seed(0)
						SCREAMING_SNAKE_CASE						= 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=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')		,					)
						SCREAMING_SNAKE_CASE						= IPNDMScheduler()

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






			def    SCREAMING_SNAKE_CASE__    (       self		,					a		,					a=0)      ->       List[str]:
						if str(a).startswith('mps'):
									SCREAMING_SNAKE_CASE						= torch.manual_seed(a)
						else:
									SCREAMING_SNAKE_CASE						= torch.Generator(device=a).manual_seed(a)
						SCREAMING_SNAKE_CASE						= {
						    'batch_size': 1,
						    'generator': generator,
						    'num_inference_steps': 4,
						}
						return inputs






			def    SCREAMING_SNAKE_CASE__    (       self)      ->       Any:
						SCREAMING_SNAKE_CASE						= 'cpu'  # ensure determinism for the device-dependent torch.Generator
						SCREAMING_SNAKE_CASE						= self.get_dummy_components()
						SCREAMING_SNAKE_CASE						= DanceDiffusionPipeline(**a)
						SCREAMING_SNAKE_CASE						= pipe.to(a)
						pipe.set_progress_bar_config(disable=a)

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

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

						assert audio.shape == (1, 2, components["unet"].sample_size)
						SCREAMING_SNAKE_CASE						= np.array([-0.72_65, 1.00_00, -0.83_88, 0.11_75, 0.94_98, -1.00_00])
						assert np.abs(audio_slice.flatten() - expected_slice).max() < 1E-2






			@skip_mps
			def    SCREAMING_SNAKE_CASE__    (       self)      ->       Any:
						return super().test_save_load_local()






			@skip_mps
			def    SCREAMING_SNAKE_CASE__    (       self)      ->       Tuple:
						return super().test_dict_tuple_outputs_equivalent(expected_max_difference=3E-3)






			@skip_mps
			def    SCREAMING_SNAKE_CASE__    (       self)      ->       Optional[int]:
						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 		_snake_case  (  unittest.TestCase					):






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






			def    SCREAMING_SNAKE_CASE__    (       self)      ->       str:
						SCREAMING_SNAKE_CASE						= torch_device

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

						SCREAMING_SNAKE_CASE						= torch.manual_seed(0)
						SCREAMING_SNAKE_CASE						= pipe(generator=a		,					num_inference_steps=100		,					audio_length_in_s=4.0_96)
						SCREAMING_SNAKE_CASE						= output.audios

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

						assert audio.shape == (1, 2, pipe.unet.sample_size)
						SCREAMING_SNAKE_CASE						= np.array([-0.01_92, -0.02_31, -0.03_18, -0.00_59, 0.00_02, -0.00_20])

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







			def    SCREAMING_SNAKE_CASE__    (       self)      ->       Optional[int]:
						SCREAMING_SNAKE_CASE						= torch_device

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

						SCREAMING_SNAKE_CASE						= torch.manual_seed(0)
						SCREAMING_SNAKE_CASE						= pipe(generator=a		,					num_inference_steps=100		,					audio_length_in_s=4.0_96)
						SCREAMING_SNAKE_CASE						= output.audios

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

						assert audio.shape == (1, 2, pipe.unet.sample_size)
						SCREAMING_SNAKE_CASE						= np.array([-0.03_67, -0.04_88, -0.07_71, -0.05_25, -0.04_44, -0.03_41])

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












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 		_snake_case  (  A__					):
			_lowercase		:    Union[str, Any]			     =							['''image_processor''', '''tokenizer''']
			_lowercase		:    Optional[int]			     =							'''Pix2StructImageProcessor'''
			_lowercase		:    str			     =							('''T5Tokenizer''', '''T5TokenizerFast''')






			def __init__(       self		,					a		,					a)      ->       Any:
						SCREAMING_SNAKE_CASE						= False
						super().__init__(a		,					a)






			def __call__(       self		,					a=None		,					a = None		,					a = True		,					a = False		,					a = None		,					a = None		,					a = 2048		,					a = 0		,					a = None		,					a = None		,					a = False		,					a = False		,					a = False		,					a = False		,					a = False		,					a = True		,					a = None		,					**a		,					)      ->       BatchEncoding:
						if images is None and text is None:
									raise ValueError('You have to specify either images or text.')

						# Get only text
						if images is None and not self.image_processor.is_vqa:
									SCREAMING_SNAKE_CASE						= self.tokenizer
									SCREAMING_SNAKE_CASE						= self.tokenizer(
									    text=a		,					add_special_tokens=a		,					padding=a		,					truncation=a		,					max_length=a		,					stride=a		,					pad_to_multiple_of=a		,					return_attention_mask=a		,					return_overflowing_tokens=a		,					return_special_tokens_mask=a		,					return_offsets_mapping=a		,					return_token_type_ids=a		,					return_length=a		,					verbose=a		,					return_tensors=a		,					**a		,					)
									return text_encoding

						if not self.image_processor.is_vqa:
									# add pixel_values
									SCREAMING_SNAKE_CASE						= self.image_processor(
									    a		,					return_tensors=a		,					max_patches=a		,					**a)
						else:
									# add pixel_values and bbox
									SCREAMING_SNAKE_CASE						= self.image_processor(
									    a		,					return_tensors=a		,					max_patches=a		,					header_text=a		,					**a)

						if text is not None and not self.image_processor.is_vqa:
									SCREAMING_SNAKE_CASE						= self.tokenizer(
									    text=a		,					add_special_tokens=a		,					padding=a		,					truncation=a		,					max_length=a		,					stride=a		,					pad_to_multiple_of=a		,					return_attention_mask=a		,					return_overflowing_tokens=a		,					return_special_tokens_mask=a		,					return_offsets_mapping=a		,					return_token_type_ids=a		,					return_length=a		,					verbose=a		,					return_tensors=a		,					**a		,					)

									if "attention_mask" in text_encoding:
												SCREAMING_SNAKE_CASE						= text_encoding.pop('attention_mask')
									if "input_ids" in text_encoding:
												SCREAMING_SNAKE_CASE						= text_encoding.pop('input_ids')
						else:
									SCREAMING_SNAKE_CASE						= None

						if text_encoding is not None:
									encoding_image_processor.update(a)

						return encoding_image_processor






			def    SCREAMING_SNAKE_CASE__    (       self		,					*a		,					**a)      ->       List[Any]:
						return self.tokenizer.batch_decode(*a		,					**a)






			def    SCREAMING_SNAKE_CASE__    (       self		,					*a		,					**a)      ->       Optional[int]:
						return self.tokenizer.decode(*a		,					**a)







			@property
			def    SCREAMING_SNAKE_CASE__    (       self)      ->       Tuple:
						SCREAMING_SNAKE_CASE						= self.tokenizer.model_input_names
						SCREAMING_SNAKE_CASE						= self.image_processor.model_input_names
						return list(dict.fromkeys(tokenizer_input_names + image_processor_input_names))




