



'''simple docstring'''






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__(      lowerCamelCase__	,				unittest.TestCase			):
    lowercase__				=					DanceDiffusionPipeline
    lowercase__				=					UNCONDITIONAL_AUDIO_GENERATION_PARAMS
    lowercase__				=					PipelineTesterMixin.required_optional_params - {
        """callback""",
        """latents""",
        """callback_steps""",
        """output_type""",
        """num_images_per_prompt""",
    }
    lowercase__				=					UNCONDITIONAL_AUDIO_GENERATION_BATCH_PARAMS
    lowercase__				=					False
    lowercase__				=					False







    def     lowercase_ (   self				:						str			):
      torch.manual_seed(0			)
      a		:      Union[str, Any]       =  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=__snake_case		,    use_timestep_embedding=__snake_case		,    time_embedding_type='fourier'		,    mid_block_type='UNetMidBlock1D'		,    down_block_types=('DownBlock1DNoSkip', 'DownBlock1D', 'AttnDownBlock1D')		,    up_block_types=('AttnUpBlock1D', 'UpBlock1D', 'UpBlock1DNoSkip')		,    )
      a		:      Optional[int]       =  IPNDMScheduler()

      a		:      Union[str, Any]       =  {
          'unet': unet,
          'scheduler': scheduler,
      }
      return components







    def     lowercase_ (   self				:						str		,    __snake_case				:						List[Any]		,    __snake_case				:						Optional[Any]=0			):
      if str(__snake_case			).startswith('mps'			):
        a		:      List[str]       =  torch.manual_seed(__snake_case			)
      else:
        a		:      List[str]       =  torch.Generator(device=__snake_case			).manual_seed(__snake_case			)
      a		:      Optional[int]       =  {
          'batch_size': 1,
          'generator': generator,
          'num_inference_steps': 4,
      }
      return inputs







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

      a		:      Any       =  self.get_dummy_inputs(__snake_case			)
      a		:      List[str]       =  pipe(**__snake_case			)
      a		:      Optional[Any]       =  output.audios

      a		:      Optional[Any]       =  audio[0, -3:, -3:]

      assert audio.shape == (1, 2, components["unet"].sample_size)
      a		:      List[str]       =  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     lowercase_ (   self				:						Optional[int]			):
      return super().test_save_load_local()







    @skip_mps
    def     lowercase_ (   self				:						str			):
      return super().test_dict_tuple_outputs_equivalent(expected_max_difference=3e-3			)







    @skip_mps
    def     lowercase_ (   self				:						Optional[int]			):
      return super().test_save_load_optional_components()







    @skip_mps
    def     lowercase_ (   self				:						Dict			):
      return super().test_attention_slicing_forward_pass()







    def     lowercase_ (   self				:						Optional[Any]			):
      super().test_inference_batch_single_identical(expected_max_diff=3e-3			)

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







    def     lowercase_ (   self				:						List[str]			):
      # clean up the VRAM after each test
      super().tearDown()
      gc.collect()
      torch.cuda.empty_cache()







    def     lowercase_ (   self				:						str			):
      a		:      Union[str, Any]       =  torch_device

      a		:      int       =  DanceDiffusionPipeline.from_pretrained('harmonai/maestro-150k'			)
      a		:      Union[str, Any]       =  pipe.to(__snake_case			)
      pipe.set_progress_bar_config(disable=__snake_case			)

      a		:      str       =  torch.manual_seed(0			)
      a		:      Any       =  pipe(generator=__snake_case		,    num_inference_steps=1_00		,    audio_length_in_s=4.096			)
      a		:      str       =  output.audios

      a		:      Tuple       =  audio[0, -3:, -3:]

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

      a		:      List[str]       =  DanceDiffusionPipeline.from_pretrained('harmonai/maestro-150k'		,    torch_dtype=torch.floataa			)
      a		:      Optional[Any]       =  pipe.to(__snake_case			)
      pipe.set_progress_bar_config(disable=__snake_case			)

      a		:      Optional[Any]       =  torch.manual_seed(0			)
      a		:      Dict       =  pipe(generator=__snake_case		,    num_inference_steps=1_00		,    audio_length_in_s=4.096			)
      a		:      Union[str, Any]       =  output.audios

      a		:      Optional[Any]       =  audio[0, -3:, -3:]

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




'''simple docstring'''






import warnings

from ...utils import logging
from .image_processing_chinese_clip import ChineseCLIPImageProcessor


lowerCAmelCase:      Optional[Any]		 =		logging.get_logger(__name__)

class 							a__(      lowerCamelCase__			):







    def __init__(   self				:						Union[str, Any]		,    *__snake_case				:						List[str]		,    **__snake_case				:						Union[str, Any]			):
      warnings.warn(
          'The class ChineseCLIPFeatureExtractor is deprecated and will be removed in version 5 of Transformers.'
          ' Please use ChineseCLIPImageProcessor instead.'		,    __snake_case		,    )
      super().__init__(*__snake_case		,    **__snake_case			)