

'''simple docstring'''






from typing import Dict, List, Optional, Union

import numpy as np

from ...image_processing_utils import BaseImageProcessor, BatchFeature, get_size_dict
from ...image_transforms import rescale, resize, to_channel_dimension_format
from ...image_utils import (
    ChannelDimension,
    ImageInput,
    PILImageResampling,
    make_list_of_images,
    to_numpy_array,
    valid_images,
)
from ...utils import TensorType, is_vision_available, logging


if is_vision_available():
 import PIL


lowerCAmelCase__   		=      logging.get_logger(__name__)


def 						_A	(       A__	,							A__ ):






 """simple docstring"""



 __lowercase				    =					b.T
 __lowercase				    =					np.sum(np.square(A__ )	,							axis=1 )
 __lowercase				    =					np.sum(np.square(A__ )	,							axis=0 )
 __lowercase				    =					np.matmul(A__	,							A__ )
 __lowercase				    =					aa[:, None] - 2 * ab + ba[None, :]
 return d


def 						_A	(       A__	,							A__ ):






 """simple docstring"""



 __lowercase				    =					x.reshape(-1	,							3 )
 __lowercase				    =					squared_euclidean_distance(A__	,							A__ )
 return np.argmin(A__	,							axis=1 )






class      lowercase_      (lowerCamelCase__ ):




  """simple docstring"""




  SCREAMING_SNAKE_CASE       :    Optional[Any]							     =     ['pixel_values']






  def __init__( self  :	Any   ,lowercase__  :	Optional[Union[List[List[int]], np.ndarray]] = None   ,lowercase__  :	bool = True   ,lowercase__  :	Dict[str, int] = None   ,lowercase__  :	PILImageResampling = PILImageResampling.BILINEAR   ,lowercase__  :	bool = True   ,lowercase__  :	bool = True   ,**lowercase__  :	Optional[Any]   ,):
   super().__init__(**lowercase__     )
   __lowercase				    =					size if size is not None else {'''height''': 2_5_6, '''width''': 2_5_6}
   __lowercase				    =					get_size_dict(lowercase__     )
   __lowercase				    =					np.array(lowercase__     ) if clusters is not None else None
   __lowercase				    =					do_resize
   __lowercase				    =					size
   __lowercase				    =					resample
   __lowercase				    =					do_normalize
   __lowercase				    =					do_color_quantize






  def    SCREAMING_SNAKE_CASE ( self  :	Tuple   ,lowercase__  :	np.ndarray   ,lowercase__  :	Dict[str, int]   ,lowercase__  :	PILImageResampling = PILImageResampling.BILINEAR   ,lowercase__  :	Optional[Union[str, ChannelDimension]] = None   ,**lowercase__  :	str   ,):
   __lowercase				    =					get_size_dict(lowercase__     )
   if "height" not in size or "width" not in size:
    raise ValueError(F"Size dictionary must contain both height and width keys. Got {size.keys()}"     )
   return resize(
       lowercase__   ,size=(size['''height'''], size['''width'''])   ,resample=lowercase__   ,data_format=lowercase__   ,**lowercase__     )






  def    SCREAMING_SNAKE_CASE ( self  :	Dict   ,lowercase__  :	np.ndarray   ,lowercase__  :	Optional[Union[str, ChannelDimension]] = None   ,):
   __lowercase				    =					rescale(image=lowercase__   ,scale=1 / 1_2_7.5   ,data_format=lowercase__     )
   __lowercase				    =					image - 1
   return image






  def    SCREAMING_SNAKE_CASE ( self  :	Dict   ,lowercase__  :	ImageInput   ,lowercase__  :	bool = None   ,lowercase__  :	Dict[str, int] = None   ,lowercase__  :	PILImageResampling = None   ,lowercase__  :	bool = None   ,lowercase__  :	Optional[bool] = None   ,lowercase__  :	Optional[Union[List[List[int]], np.ndarray]] = None   ,lowercase__  :	Optional[Union[str, TensorType]] = None   ,lowercase__  :	Optional[Union[str, ChannelDimension]] = ChannelDimension.FIRST   ,**lowercase__  :	Any   ,):
   __lowercase				    =					do_resize if do_resize is not None else self.do_resize
   __lowercase				    =					size if size is not None else self.size
   __lowercase				    =					get_size_dict(lowercase__     )
   __lowercase				    =					resample if resample is not None else self.resample
   __lowercase				    =					do_normalize if do_normalize is not None else self.do_normalize
   __lowercase				    =					do_color_quantize if do_color_quantize is not None else self.do_color_quantize
   __lowercase				    =					clusters if clusters is not None else self.clusters
   __lowercase				    =					np.array(lowercase__     )

   __lowercase				    =					make_list_of_images(lowercase__     )

   if not valid_images(lowercase__     ):
    raise ValueError(
        '''Invalid image type. Must be of type PIL.Image.Image, numpy.ndarray, '''
        '''torch.Tensor, tf.Tensor or jax.ndarray.'''     )

   if do_resize and size is None or resample is None:
    raise ValueError('''Size and resample must be specified if do_resize is True.'''     )

   if do_color_quantize and clusters is None:
    raise ValueError('''Clusters must be specified if do_color_quantize is True.'''     )

   # All transformations expect numpy arrays.
   __lowercase				    =					[to_numpy_array(lowercase__     ) for image in images]

   if do_resize:
    __lowercase				    =					[self.resize(image=lowercase__   ,size=lowercase__   ,resample=lowercase__     ) for image in images]

   if do_normalize:
    __lowercase				    =					[self.normalize(image=lowercase__     ) for image in images]

   if do_color_quantize:
    __lowercase				    =					[to_channel_dimension_format(lowercase__   ,ChannelDimension.LAST     ) for image in images]
    # color quantize from (batch_size, height, width, 3) to (batch_size, height, width)
    __lowercase				    =					np.array(lowercase__     )
    __lowercase				    =					color_quantize(lowercase__   ,lowercase__     ).reshape(images.shape[:-1]     )

    # flatten to (batch_size, height*width)
    __lowercase				    =					images.shape[0]
    __lowercase				    =					images.reshape(lowercase__   ,-1     )

    # We need to convert back to a list of images to keep consistent behaviour across processors.
    __lowercase				    =					list(lowercase__     )
   else:
    __lowercase				    =					[to_channel_dimension_format(lowercase__   ,lowercase__     ) for image in images]

   __lowercase				    =					{'''input_ids''': images}
   return BatchFeature(data=lowercase__   ,tensor_type=lowercase__     )



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			):





     """simple docstring"""







     SCREAMING_SNAKE_CASE_		 =	DanceDiffusionPipeline
     SCREAMING_SNAKE_CASE_		 =	UNCONDITIONAL_AUDIO_GENERATION_PARAMS
     SCREAMING_SNAKE_CASE_		 =	PipelineTesterMixin.required_optional_params - {
         '''callback''',
         '''latents''',
         '''callback_steps''',
         '''output_type''',
         '''num_images_per_prompt''',
     }
     SCREAMING_SNAKE_CASE_		 =	UNCONDITIONAL_AUDIO_GENERATION_BATCH_PARAMS
     SCREAMING_SNAKE_CASE_		 =	False
     SCREAMING_SNAKE_CASE_		 =	False






     def  __SCREAMING_SNAKE_CASE      (				self	)	-> Any:




           """simple docstring"""







           torch.manual_seed(0	)
           UpperCamelCase__     : List[str]											=					UNetaDModel(
               block_out_channels=(3_2, 3_2, 6_4)    ,    extra_in_channels=1_6    ,    sample_size=5_1_2    ,    sample_rate=1_6_0_0_0    ,    in_channels=2    ,    out_channels=2    ,    flip_sin_to_cos=__SCREAMING_SNAKE_CASE    ,    use_timestep_embedding=__SCREAMING_SNAKE_CASE    ,    time_embedding_type='''fourier'''    ,    mid_block_type='''UNetMidBlock1D'''    ,    down_block_types=('''DownBlock1DNoSkip''', '''DownBlock1D''', '''AttnDownBlock1D''')    ,    up_block_types=('''AttnUpBlock1D''', '''UpBlock1D''', '''UpBlock1DNoSkip''')    ,    )
           UpperCamelCase__     : Union[str, Any]											=					IPNDMScheduler()

           UpperCamelCase__     : List[str]											=					{
               '''unet''': unet,
               '''scheduler''': scheduler,
           }
           return components






     def  __SCREAMING_SNAKE_CASE      (				self    ,    __SCREAMING_SNAKE_CASE    ,    __SCREAMING_SNAKE_CASE=0	)	-> Any:




           """simple docstring"""







           if str(__SCREAMING_SNAKE_CASE	).startswith('''mps'''	):
                 UpperCamelCase__     : Optional[Any]											=					torch.manual_seed(__SCREAMING_SNAKE_CASE	)
           else:
                 UpperCamelCase__     : Optional[Any]											=					torch.Generator(device=__SCREAMING_SNAKE_CASE	).manual_seed(__SCREAMING_SNAKE_CASE	)
           UpperCamelCase__     : int											=					{
               '''batch_size''': 1,
               '''generator''': generator,
               '''num_inference_steps''': 4,
           }
           return inputs






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




           """simple docstring"""







           UpperCamelCase__     : str											=					'''cpu'''  # ensure determinism for the device-dependent torch.Generator
           UpperCamelCase__     : Dict											=					self.get_dummy_components()
           UpperCamelCase__     : str											=					DanceDiffusionPipeline(**__SCREAMING_SNAKE_CASE	)
           UpperCamelCase__     : List[Any]											=					pipe.to(__SCREAMING_SNAKE_CASE	)
           pipe.set_progress_bar_config(disable=__SCREAMING_SNAKE_CASE	)

           UpperCamelCase__     : int											=					self.get_dummy_inputs(__SCREAMING_SNAKE_CASE	)
           UpperCamelCase__     : Dict											=					pipe(**__SCREAMING_SNAKE_CASE	)
           UpperCamelCase__     : List[str]											=					output.audios

           UpperCamelCase__     : Dict											=					audio[0, -3:, -3:]

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




           """simple docstring"""







           return super().test_save_load_local()






     @skip_mps
     def  __SCREAMING_SNAKE_CASE      (				self	)	-> Tuple:




           """simple docstring"""







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






     @skip_mps
     def  __SCREAMING_SNAKE_CASE      (				self	)	-> str:




           """simple docstring"""







           return super().test_save_load_optional_components()






     @skip_mps
     def  __SCREAMING_SNAKE_CASE      (				self	)	-> Optional[int]:




           """simple docstring"""







           return super().test_attention_slicing_forward_pass()






     def  __SCREAMING_SNAKE_CASE      (				self	)	-> Tuple:




           """simple docstring"""







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




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





     """simple docstring"""







     def  __SCREAMING_SNAKE_CASE      (				self	)	-> int:




           """simple docstring"""







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






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




           """simple docstring"""







           UpperCamelCase__     : Any											=					torch_device

           UpperCamelCase__     : Any											=					DanceDiffusionPipeline.from_pretrained('''harmonai/maestro-150k'''	)
           UpperCamelCase__     : int											=					pipe.to(__SCREAMING_SNAKE_CASE	)
           pipe.set_progress_bar_config(disable=__SCREAMING_SNAKE_CASE	)

           UpperCamelCase__     : int											=					torch.manual_seed(0	)
           UpperCamelCase__     : Optional[Any]											=					pipe(generator=__SCREAMING_SNAKE_CASE    ,    num_inference_steps=1_0_0    ,    audio_length_in_s=4.096	)
           UpperCamelCase__     : str											=					output.audios

           UpperCamelCase__     : List[Any]											=					audio[0, -3:, -3:]

           assert audio.shape == (1, 2, pipe.unet.sample_size)
           UpperCamelCase__     : Tuple											=					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  __SCREAMING_SNAKE_CASE      (				self	)	-> Union[str, Any]:




           """simple docstring"""







           UpperCamelCase__     : Any											=					torch_device

           UpperCamelCase__     : Union[str, Any]											=					DanceDiffusionPipeline.from_pretrained('''harmonai/maestro-150k'''    ,    torch_dtype=torch.floataa	)
           UpperCamelCase__     : Tuple											=					pipe.to(__SCREAMING_SNAKE_CASE	)
           pipe.set_progress_bar_config(disable=__SCREAMING_SNAKE_CASE	)

           UpperCamelCase__     : int											=					torch.manual_seed(0	)
           UpperCamelCase__     : Union[str, Any]											=					pipe(generator=__SCREAMING_SNAKE_CASE    ,    num_inference_steps=1_0_0    ,    audio_length_in_s=4.096	)
           UpperCamelCase__     : List[Any]											=					output.audios

           UpperCamelCase__     : List[Any]											=					audio[0, -3:, -3:]

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


