
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__						     =      DanceDiffusionPipeline
							UpperCAmelCase__						     =      UNCONDITIONAL_AUDIO_GENERATION_PARAMS
							UpperCAmelCase__						     =      PipelineTesterMixin.required_optional_params - {
							    '''callback''',
							    '''latents''',
							    '''callback_steps''',
							    '''output_type''',
							    '''num_images_per_prompt''',
							}
							UpperCAmelCase__						     =      UNCONDITIONAL_AUDIO_GENERATION_BATCH_PARAMS
							UpperCAmelCase__						     =      False
							UpperCAmelCase__						     =      False

							def _lowercase					(self    ):

									"""simple docstring"""







									torch.manual_seed(0    )
									SCREAMING_SNAKE_CASE_						 = 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=__UpperCamelCase       ,       use_timestep_embedding=__UpperCamelCase       ,       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 _lowercase					(self       ,       SCREAMING_SNAKE_CASE_       ,       SCREAMING_SNAKE_CASE_=0    ):

									"""simple docstring"""







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

							def _lowercase					(self    ):

									"""simple docstring"""







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

									SCREAMING_SNAKE_CASE_						 = self.get_dummy_inputs(__UpperCamelCase    )
									SCREAMING_SNAKE_CASE_						 = pipe(**__UpperCamelCase    )
									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 _lowercase					(self    ):

									"""simple docstring"""







									return super().test_save_load_local()

							@skip_mps
							def _lowercase					(self    ):

									"""simple docstring"""







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

							@skip_mps
							def _lowercase					(self    ):

									"""simple docstring"""







									return super().test_save_load_optional_components()

							@skip_mps
							def _lowercase					(self    ):

									"""simple docstring"""







									return super().test_attention_slicing_forward_pass()







							def _lowercase					(self    ):

									"""simple docstring"""







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







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

							def _lowercase					(self    ):

									"""simple docstring"""







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

							def _lowercase					(self    ):

									"""simple docstring"""







									SCREAMING_SNAKE_CASE_						 = torch_device

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

									SCREAMING_SNAKE_CASE_						 = torch.manual_seed(0    )
									SCREAMING_SNAKE_CASE_						 = pipe(generator=__UpperCamelCase       ,       num_inference_steps=1_00       ,       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 _lowercase					(self    ):

									"""simple docstring"""







									SCREAMING_SNAKE_CASE_						 = torch_device

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

									SCREAMING_SNAKE_CASE_						 = torch.manual_seed(0    )
									SCREAMING_SNAKE_CASE_						 = pipe(generator=__UpperCamelCase       ,       num_inference_steps=1_00       ,       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

"""simple docstring"""




from dataclasses import dataclass
from typing import Optional, Tuple

import torch
from torch import nn
from transformers import RobertaPreTrainedModel, XLMRobertaConfig, XLMRobertaModel
from transformers.utils import ModelOutput





@dataclass
class  snake_case     (							__lowercase   ):
							UpperCAmelCase__						     =      None
							UpperCAmelCase__						     =      None
							UpperCAmelCase__						     =      None
							UpperCAmelCase__						     =      None







class  snake_case     (							__lowercase   ):

							def __init__(self       ,       SCREAMING_SNAKE_CASE_=1       ,       SCREAMING_SNAKE_CASE_=0       ,       SCREAMING_SNAKE_CASE_=2       ,       SCREAMING_SNAKE_CASE_=5_12       ,       SCREAMING_SNAKE_CASE_="cls"       ,       SCREAMING_SNAKE_CASE_=False       ,       SCREAMING_SNAKE_CASE_=True       ,       **SCREAMING_SNAKE_CASE_       ,       ):

									"""simple docstring"""







									super().__init__(pad_token_id=SCREAMING_SNAKE_CASE_       ,       bos_token_id=SCREAMING_SNAKE_CASE_       ,       eos_token_id=SCREAMING_SNAKE_CASE_       ,       **SCREAMING_SNAKE_CASE_    )
									SCREAMING_SNAKE_CASE_						 = project_dim
									SCREAMING_SNAKE_CASE_						 = pooler_fn
									SCREAMING_SNAKE_CASE_						 = learn_encoder
									SCREAMING_SNAKE_CASE_						 = use_attention_mask







class  snake_case     (							__lowercase   ):
							UpperCAmelCase__						     =      [R'''pooler''', R'''logit_scale''']
							UpperCAmelCase__						     =      [R'''position_ids''', R'''predictions.decoder.bias''']
							UpperCAmelCase__						     =      '''roberta'''
							UpperCAmelCase__						     =      RobertaSeriesConfig

							def __init__(self       ,       SCREAMING_SNAKE_CASE_    ):

									"""simple docstring"""







									super().__init__(SCREAMING_SNAKE_CASE_    )
									SCREAMING_SNAKE_CASE_						 = XLMRobertaModel(SCREAMING_SNAKE_CASE_    )
									SCREAMING_SNAKE_CASE_						 = nn.Linear(config.hidden_size       ,       config.project_dim    )
									SCREAMING_SNAKE_CASE_						 = getattr(SCREAMING_SNAKE_CASE_       ,       '''has_pre_transformation'''       ,       SCREAMING_SNAKE_CASE_    )
									if self.has_pre_transformation:
											SCREAMING_SNAKE_CASE_						 = nn.Linear(config.hidden_size       ,       config.project_dim    )
											SCREAMING_SNAKE_CASE_						 = nn.LayerNorm(config.hidden_size       ,       eps=config.layer_norm_eps    )
									self.post_init()







							def _lowercase					(self       ,       SCREAMING_SNAKE_CASE_ = None       ,       SCREAMING_SNAKE_CASE_ = None       ,       SCREAMING_SNAKE_CASE_ = None       ,       SCREAMING_SNAKE_CASE_ = None       ,       SCREAMING_SNAKE_CASE_ = None       ,       SCREAMING_SNAKE_CASE_ = None       ,       SCREAMING_SNAKE_CASE_ = None       ,       SCREAMING_SNAKE_CASE_ = None       ,       SCREAMING_SNAKE_CASE_ = None       ,       SCREAMING_SNAKE_CASE_ = None       ,       SCREAMING_SNAKE_CASE_ = None       ,       ):

									"""simple docstring"""







									SCREAMING_SNAKE_CASE_						 = return_dict if return_dict is not None else self.config.use_return_dict

									SCREAMING_SNAKE_CASE_						 = self.base_model(
									    input_ids=SCREAMING_SNAKE_CASE_       ,       attention_mask=SCREAMING_SNAKE_CASE_       ,       token_type_ids=SCREAMING_SNAKE_CASE_       ,       position_ids=SCREAMING_SNAKE_CASE_       ,       head_mask=SCREAMING_SNAKE_CASE_       ,       inputs_embeds=SCREAMING_SNAKE_CASE_       ,       encoder_hidden_states=SCREAMING_SNAKE_CASE_       ,       encoder_attention_mask=SCREAMING_SNAKE_CASE_       ,       output_attentions=SCREAMING_SNAKE_CASE_       ,       output_hidden_states=True if self.has_pre_transformation else output_hidden_states       ,       return_dict=SCREAMING_SNAKE_CASE_       ,       )

									if self.has_pre_transformation:
											SCREAMING_SNAKE_CASE_						 = outputs['''hidden_states'''][-2]
											SCREAMING_SNAKE_CASE_						 = self.pre_LN(SCREAMING_SNAKE_CASE_    )
											SCREAMING_SNAKE_CASE_						 = self.transformation_pre(SCREAMING_SNAKE_CASE_    )

											return TransformationModelOutput(
											    projection_state=SCREAMING_SNAKE_CASE_       ,       last_hidden_state=outputs.last_hidden_state       ,       hidden_states=outputs.hidden_states       ,       attentions=outputs.attentions       ,       )
									else:
											SCREAMING_SNAKE_CASE_						 = self.transformation(outputs.last_hidden_state    )
											return TransformationModelOutput(
											    projection_state=SCREAMING_SNAKE_CASE_       ,       last_hidden_state=outputs.last_hidden_state       ,       hidden_states=outputs.hidden_states       ,       attentions=outputs.attentions       ,       )