"""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  snake_case__      (       UpperCamelCase_,						unittest.TestCase       ):
 _snake_case       :					int							  =						DanceDiffusionPipeline
 _snake_case       :					Optional[Any]							  =						UNCONDITIONAL_AUDIO_GENERATION_PARAMS
 _snake_case       :					Optional[Any]							  =						PipelineTesterMixin.required_optional_params - {
     """callback""",
     """latents""",
     """callback_steps""",
     """output_type""",
     """num_images_per_prompt""",
 }
 _snake_case       :					Optional[int]							  =						UNCONDITIONAL_AUDIO_GENERATION_BATCH_PARAMS
 _snake_case       :					List[Any]							  =						False
 _snake_case       :					Any							  =						False


 def 						a__   ( self						):
        torch.manual_seed(0						)
        __a       =    UNetaDModel(
            block_out_channels=(32, 32, 64)    ,		extra_in_channels=16    ,		sample_size=512    ,		sample_rate=16000    ,		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")    ,		)
        __a       =    IPNDMScheduler()

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


 def 						a__   ( self    ,		lowerCamelCase    ,		lowerCamelCase=0						):







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


 def 						a__   ( self						):
        __a       =    '''cpu'''  # ensure determinism for the device-dependent torch.Generator
        __a       =    self.get_dummy_components()
        __a       =    DanceDiffusionPipeline(**__SCREAMING_SNAKE_CASE						)
        __a       =    pipe.to(__SCREAMING_SNAKE_CASE						)
        pipe.set_progress_bar_config(disable=__SCREAMING_SNAKE_CASE						)

        __a       =    self.get_dummy_inputs(__SCREAMING_SNAKE_CASE						)
        __a       =    pipe(**__SCREAMING_SNAKE_CASE						)
        __a       =    output.audios

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

        assert audio.shape == (1, 2, components["unet"].sample_size)
        __a       =    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 						a__   ( self						):
        return super().test_save_load_local()


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


 @skip_mps
 def 						a__   ( self						):
        return super().test_save_load_optional_components()


 @skip_mps
 def 						a__   ( self						):
        return super().test_attention_slicing_forward_pass()






 def 						a__   ( self						):
        super().test_inference_batch_single_identical(expected_max_diff=3E-3						)






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


 def 						a__   ( self						):
        super().tearDown()
        gc.collect()
        torch.cuda.empty_cache()


 def 						a__   ( self						):
        __a       =    torch_device

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

        __a       =    torch.manual_seed(0						)
        __a       =    pipe(generator=__SCREAMING_SNAKE_CASE    ,		num_inference_steps=100    ,		audio_length_in_s=4.096						)
        __a       =    output.audios

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

        assert audio.shape == (1, 2, pipe.unet.sample_size)
        __a       =    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 						a__   ( self						):
        __a       =    torch_device

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

        __a       =    torch.manual_seed(0						)
        __a       =    pipe(generator=__SCREAMING_SNAKE_CASE    ,		num_inference_steps=100    ,		audio_length_in_s=4.096						)
        __a       =    output.audios

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

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


from typing import List, Union

from ..utils import (
    add_end_docstrings,
    is_tf_available,
    is_torch_available,
    is_vision_available,
    logging,
    requires_backends,
)
from .base import PIPELINE_INIT_ARGS, Pipeline


if is_vision_available():
    from PIL import Image

    from ..image_utils import load_image

if is_tf_available():
    from ..models.auto.modeling_tf_auto import TF_MODEL_FOR_VISION_2_SEQ_MAPPING

if is_torch_available():
    import torch

    from ..models.auto.modeling_auto import MODEL_FOR_VISION_2_SEQ_MAPPING

lowerCamelCase  =logging.get_logger(__name__)




@add_end_docstrings(UpperCamelCase_			)
class 					_lowerCamelCase			(   UpperCamelCase_			):





     """simple docstring"""







     def __init__(				self    ,    *__SCREAMING_SNAKE_CASE    ,    **__SCREAMING_SNAKE_CASE	)	-> Dict:




           """simple docstring"""







           super().__init__(*__SCREAMING_SNAKE_CASE    ,    **__SCREAMING_SNAKE_CASE	)
           requires_backends(self    ,    '''vision'''	)
           self.check_model_type(
               TF_MODEL_FOR_VISION_2_SEQ_MAPPING if self.framework == '''tf''' else MODEL_FOR_VISION_2_SEQ_MAPPING	)






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




           """simple docstring"""







           UpperCamelCase__     : Tuple											=					{}
           UpperCamelCase__     : int											=					{}

           if prompt is not None:
                 UpperCamelCase__     : int											=					prompt

           if generate_kwargs is not None:
                 UpperCamelCase__     : int											=					generate_kwargs
           if max_new_tokens is not None:
                 if "generate_kwargs" not in forward_kwargs:
                       UpperCamelCase__     : Union[str, Any]											=					{}
                 if "max_new_tokens" in forward_kwargs["generate_kwargs"]:
                       raise ValueError(
                           '''\'max_new_tokens\' is defined twice, once in \'generate_kwargs\' and once as a direct parameter,'''
                           ''' please use only one'''	)
                 UpperCamelCase__     : Dict											=					max_new_tokens
           return preprocess_params, forward_kwargs, {}






     def __call__(				self    ,    __SCREAMING_SNAKE_CASE    ,    **__SCREAMING_SNAKE_CASE	)	-> List[str]:




           """simple docstring"""







           return super().__call__(__SCREAMING_SNAKE_CASE    ,    **__SCREAMING_SNAKE_CASE	)






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




           """simple docstring"""







           UpperCamelCase__     : List[Any]											=					load_image(__SCREAMING_SNAKE_CASE	)

           if prompt is not None:
                 if not isinstance(__SCREAMING_SNAKE_CASE    ,    __SCREAMING_SNAKE_CASE	):
                       raise ValueError(
                           F'''Received an invalid text input, got - {type(__SCREAMING_SNAKE_CASE	)} - but expected a single string. '''
                           '''Note also that one single text can be provided for conditional image to text generation.'''	)

                 UpperCamelCase__     : Optional[int]											=					self.model.config.model_type

                 if model_type == "git":
                       UpperCamelCase__     : Optional[int]											=					self.image_processor(images=__SCREAMING_SNAKE_CASE    ,    return_tensors=self.framework	)
                       UpperCamelCase__     : str											=					self.tokenizer(text=__SCREAMING_SNAKE_CASE    ,    add_special_tokens=__SCREAMING_SNAKE_CASE	).input_ids
                       UpperCamelCase__     : Dict											=					[self.tokenizer.cls_token_id] + input_ids
                       UpperCamelCase__     : int											=					torch.tensor(__SCREAMING_SNAKE_CASE	).unsqueeze(0	)
                       model_inputs.update({'''input_ids''': input_ids}	)

                 elif model_type == "pix2struct":
                       UpperCamelCase__     : Tuple											=					self.image_processor(images=__SCREAMING_SNAKE_CASE    ,    header_text=__SCREAMING_SNAKE_CASE    ,    return_tensors=self.framework	)

                 elif model_type != "vision-encoder-decoder":
                       # vision-encoder-decoder does not support conditional generation
                       UpperCamelCase__     : int											=					self.image_processor(images=__SCREAMING_SNAKE_CASE    ,    return_tensors=self.framework	)
                       UpperCamelCase__     : int											=					self.tokenizer(__SCREAMING_SNAKE_CASE    ,    return_tensors=self.framework	)
                       model_inputs.update(__SCREAMING_SNAKE_CASE	)

                 else:
                       raise ValueError(F'''Model type {model_type} does not support conditional text generation'''	)

           else:
                 UpperCamelCase__     : str											=					self.image_processor(images=__SCREAMING_SNAKE_CASE    ,    return_tensors=self.framework	)

           if self.model.config.model_type == "git" and prompt is None:
                 UpperCamelCase__     : Optional[Any]											=					None

           return model_inputs






     def  __SCREAMING_SNAKE_CASE      (				self    ,    __SCREAMING_SNAKE_CASE    ,    __SCREAMING_SNAKE_CASE=None	)	-> Optional[Any]:




           """simple docstring"""







           if (
               "input_ids" in model_inputs
               and isinstance(model_inputs['''input_ids''']    ,    __SCREAMING_SNAKE_CASE	)
               and all(x is None for x in model_inputs['''input_ids''']	)
           ):
                 UpperCamelCase__     : Union[str, Any]											=					None

           if generate_kwargs is None:
                 UpperCamelCase__     : Optional[Any]											=					{}
           # FIXME: We need to pop here due to a difference in how `generation.py` and `generation.tf_utils.py`
           #  parse inputs. In the Tensorflow version, `generate` raises an error if we don't use `input_ids` whereas
           #  the PyTorch version matches it with `self.model.main_input_name` or `self.model.encoder.main_input_name`
           #  in the `_prepare_model_inputs` method.
           UpperCamelCase__     : Any											=					model_inputs.pop(self.model.main_input_name	)
           UpperCamelCase__     : Any											=					self.model.generate(__SCREAMING_SNAKE_CASE    ,    **__SCREAMING_SNAKE_CASE    ,    **__SCREAMING_SNAKE_CASE	)
           return model_outputs






     def  __SCREAMING_SNAKE_CASE      (				self    ,    __SCREAMING_SNAKE_CASE	)	-> int:




           """simple docstring"""







           UpperCamelCase__     : Union[str, Any]											=					[]
           for output_ids in model_outputs:
                 UpperCamelCase__     : str											=					{
                     '''generated_text''': self.tokenizer.decode(
                         __SCREAMING_SNAKE_CASE    ,    skip_special_tokens=__SCREAMING_SNAKE_CASE    ,    )
                 }
                 records.append(__SCREAMING_SNAKE_CASE	)
           return records


