





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     A     (SCREAMING_SNAKE_CASE  ,	unittest.TestCase       ):
			'''simple docstring'''




			__lowerCamelCase      :			Dict							     =       DanceDiffusionPipeline
			__lowerCamelCase      :			Any							     =       UNCONDITIONAL_AUDIO_GENERATION_PARAMS
			__lowerCamelCase      :			List[Any]							     =       PipelineTesterMixin.required_optional_params - {
			    '''callback''',
			    '''latents''',
			    '''callback_steps''',
			    '''output_type''',
			    '''num_images_per_prompt''',
			}
			__lowerCamelCase      :			List[Any]							     =       UNCONDITIONAL_AUDIO_GENERATION_BATCH_PARAMS
			__lowerCamelCase      :			List[str]							     =       False
			__lowerCamelCase      :			Union[str, Any]							     =       False






			def   a_			(     self  :       Optional[int]  )      ->      List[Any]:







					"""simple docstring"""






					torch.manual_seed(0  )
					A__     =  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=__lowerCAmelCase				,      use_timestep_embedding=__lowerCAmelCase				,      time_embedding_type="""fourier"""				,      mid_block_type="""UNetMidBlock1D"""				,      down_block_types=("""DownBlock1DNoSkip""", """DownBlock1D""", """AttnDownBlock1D""")				,      up_block_types=("""AttnUpBlock1D""", """UpBlock1D""", """UpBlock1DNoSkip""")				,      )
					A__     =  IPNDMScheduler()

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






			def   a_			(     self  :       List[str]				,      __lowerCAmelCase  :       Union[str, Any]				,      __lowerCAmelCase  :       List[Any]=0  )      ->      List[Any]:







					"""simple docstring"""






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






			def   a_			(     self  :       str  )      ->      str:







					"""simple docstring"""






					A__     =  """cpu"""  # ensure determinism for the device-dependent torch.Generator
					A__     =  self.get_dummy_components()
					A__     =  DanceDiffusionPipeline(**__lowerCAmelCase  )
					A__     =  pipe.to(__lowerCAmelCase  )
					pipe.set_progress_bar_config(disable=__lowerCAmelCase  )

					A__     =  self.get_dummy_inputs(__lowerCAmelCase  )
					A__     =  pipe(**__lowerCAmelCase  )
					A__     =  output.audios

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

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







					"""simple docstring"""






					return super().test_save_load_local()






			@skip_mps
			def   a_			(     self  :       List[str]  )      ->      Optional[Any]:







					"""simple docstring"""






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






			@skip_mps
			def   a_			(     self  :       Tuple  )      ->      Any:







					"""simple docstring"""






					return super().test_save_load_optional_components()






			@skip_mps
			def   a_			(     self  :       List[Any]  )      ->      Union[str, Any]:







					"""simple docstring"""






					return super().test_attention_slicing_forward_pass()







			def   a_			(     self  :       List[str]  )      ->      Optional[int]:







					"""simple docstring"""






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


@slow
@require_torch_gpu
class     A     (unittest.TestCase       ):
			'''simple docstring'''




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







					"""simple docstring"""






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






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







					"""simple docstring"""






					A__     =  torch_device

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

					A__     =  torch.manual_seed(0  )
					A__     =  pipe(generator=__lowerCAmelCase				,      num_inference_steps=1_00				,      audio_length_in_s=4.0_9_6  )
					A__     =  output.audios

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

					assert audio.shape == (1, 2, pipe.unet.sample_size)
					A__     =  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   a_			(     self  :       Dict  )      ->      Optional[int]:







					"""simple docstring"""






					A__     =  torch_device

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

					A__     =  torch.manual_seed(0  )
					A__     =  pipe(generator=__lowerCAmelCase				,      num_inference_steps=1_00				,      audio_length_in_s=4.0_9_6  )
					A__     =  output.audios

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

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

from transformers import CLIPImageProcessor, CLIPVisionModelWithProjection

from diffusers import UnCLIPImageVariationPipeline, UnCLIPPipeline


if __name__ == "__main__":
					A       :   List[Any]					= argparse.ArgumentParser()

					parser.add_argument('''--dump_path''', default=None, type=str, required=True, help='''Path to the output model.''')

					parser.add_argument(
					    '''--txt2img_unclip''',
					    default='''kakaobrain/karlo-v1-alpha''',
					    type=str,
					    required=False,
					    help='''The pretrained txt2img unclip.''',
					)

					A       :   List[Any]					= parser.parse_args()

					A       :   int					= UnCLIPPipeline.from_pretrained(args.txtaimg_unclip)

					A       :   Tuple					= CLIPImageProcessor()
					A       :   int					= CLIPVisionModelWithProjection.from_pretrained('''openai/clip-vit-large-patch14''')

					A       :   Tuple					= UnCLIPImageVariationPipeline(
					    decoder=txtaimg.decoder,
					    text_encoder=txtaimg.text_encoder,
					    tokenizer=txtaimg.tokenizer,
					    text_proj=txtaimg.text_proj,
					    feature_extractor=feature_extractor,
					    image_encoder=image_encoder,
					    super_res_first=txtaimg.super_res_first,
					    super_res_last=txtaimg.super_res_last,
					    decoder_scheduler=txtaimg.decoder_scheduler,
					    super_res_scheduler=txtaimg.super_res_scheduler,
					)

					imgaimg.save_pretrained(args.dump_path)



