





import warnings
from typing import List

import numpy as np

from ...processing_utils import ProcessorMixin
from ...tokenization_utils_base import BatchEncoding
from ...utils import is_flax_available, is_tf_available, is_torch_available



class     _UpperCAmelCase   (   A__   ):

  """simple docstring"""


  a_      =			["""image_processor""", """tokenizer"""]
  a_      =			"""OwlViTImageProcessor"""
  a_      =			("""CLIPTokenizer""", """CLIPTokenizerFast""")






  def __init__(    self			:		Tuple ,  lowerCAmelCase_			:		Optional[int]=None ,  lowerCAmelCase_			:		Optional[int]=None ,  **lowerCAmelCase_			:		int					)							->						List[str]:
     __lowerCAmelCase							     =     None
     if "feature_extractor" in kwargs:
        warnings.warn(
            'The `feature_extractor` argument is deprecated and will be removed in v5, use `image_processor`'
            ' instead.' ,  __snake_case ,  )
        __lowerCAmelCase							     =     kwargs.pop('feature_extractor'					)

     __lowerCAmelCase							     =     image_processor if image_processor is not None else feature_extractor
     if image_processor is None:
        raise ValueError('You need to specify an `image_processor`.'					)
     if tokenizer is None:
        raise ValueError('You need to specify a `tokenizer`.'					)

     super().__init__(__snake_case ,  __snake_case					)






  def __call__(    self			:		List[Any] ,  lowerCAmelCase_			:		int=None ,  lowerCAmelCase_			:		List[str]=None ,  lowerCAmelCase_			:		List[Any]=None ,  lowerCAmelCase_			:		Any="max_length" ,  lowerCAmelCase_			:		List[str]="np" ,  **lowerCAmelCase_			:		int					)							->						Any:

     if text is None and query_images is None and images is None:
        raise ValueError(
            'You have to specify at least one text or query image or image. All three cannot be none.'					)

     if text is not None:
        if isinstance(__snake_case ,  __snake_case					) or (isinstance(__snake_case ,  __snake_case					) and not isinstance(text[0] ,  __snake_case					)):
           __lowerCAmelCase							     =     [self.tokenizer(__snake_case ,  padding=__snake_case ,  return_tensors=__snake_case ,  **__snake_case					)]

        elif isinstance(__snake_case ,  __snake_case					) and isinstance(text[0] ,  __snake_case					):
           __lowerCAmelCase							     =     []

           # Maximum number of queries across batch
           __lowerCAmelCase							     =     max([len(__snake_case					) for t in text]					)

           # Pad all batch samples to max number of text queries
           for t in text:
              if len(__snake_case					) != max_num_queries:
                 __lowerCAmelCase							     =     t + [' '] * (max_num_queries - len(__snake_case					))

              __lowerCAmelCase							     =     self.tokenizer(__snake_case ,  padding=__snake_case ,  return_tensors=__snake_case ,  **__snake_case					)
              encodings.append(__snake_case					)
        else:
           raise TypeError('Input text should be a string, a list of strings or a nested list of strings'					)

        if return_tensors == "np":
           __lowerCAmelCase							     =     np.concatenate([encoding['input_ids'] for encoding in encodings] ,  axis=0					)
           __lowerCAmelCase							     =     np.concatenate([encoding['attention_mask'] for encoding in encodings] ,  axis=0					)

        elif return_tensors == "jax" and is_flax_available():
           import jax.numpy as jnp

           __lowerCAmelCase							     =     jnp.concatenate([encoding['input_ids'] for encoding in encodings] ,  axis=0					)
           __lowerCAmelCase							     =     jnp.concatenate([encoding['attention_mask'] for encoding in encodings] ,  axis=0					)

        elif return_tensors == "pt" and is_torch_available():
           import torch

           __lowerCAmelCase							     =     torch.cat([encoding['input_ids'] for encoding in encodings] ,  dim=0					)
           __lowerCAmelCase							     =     torch.cat([encoding['attention_mask'] for encoding in encodings] ,  dim=0					)

        elif return_tensors == "tf" and is_tf_available():
           import tensorflow as tf

           __lowerCAmelCase							     =     tf.stack([encoding['input_ids'] for encoding in encodings] ,  axis=0					)
           __lowerCAmelCase							     =     tf.stack([encoding['attention_mask'] for encoding in encodings] ,  axis=0					)

        else:
           raise ValueError('Target return tensor type could not be returned'					)

        __lowerCAmelCase							     =     BatchEncoding()
        __lowerCAmelCase							     =     input_ids
        __lowerCAmelCase							     =     attention_mask

     if query_images is not None:
        __lowerCAmelCase							     =     BatchEncoding()
        __lowerCAmelCase							     =     self.image_processor(
            __snake_case ,  return_tensors=__snake_case ,  **__snake_case					).pixel_values
        __lowerCAmelCase							     =     query_pixel_values

     if images is not None:
        __lowerCAmelCase							     =     self.image_processor(__snake_case ,  return_tensors=__snake_case ,  **__snake_case					)

     if text is not None and images is not None:
        __lowerCAmelCase							     =     image_features.pixel_values
        return encoding
     elif query_images is not None and images is not None:
        __lowerCAmelCase							     =     image_features.pixel_values
        return encoding
     elif text is not None or query_images is not None:
        return encoding
     else:
        return BatchEncoding(data=dict(**__snake_case					) ,  tensor_type=__snake_case					)






  def     lowercase		(    self			:		List[Any] ,  *lowerCAmelCase_			:		int ,  **lowerCAmelCase_			:		Union[str, Any]					)							->						str:
     return self.image_processor.post_process(*__snake_case ,  **__snake_case					)






  def     lowercase		(    self			:		Dict ,  *lowerCAmelCase_			:		Union[str, Any] ,  **lowerCAmelCase_			:		Tuple					)							->						List[str]:
     return self.image_processor.post_process_object_detection(*__snake_case ,  **__snake_case					)






  def     lowercase		(    self			:		List[str] ,  *lowerCAmelCase_			:		str ,  **lowerCAmelCase_			:		Union[str, Any]					)							->						Optional[int]:
     return self.image_processor.post_process_image_guided_detection(*__snake_case ,  **__snake_case					)






  def     lowercase		(    self			:		Optional[int] ,  *lowerCAmelCase_			:		List[Any] ,  **lowerCAmelCase_			:		Union[str, Any]					)							->						Optional[Any]:
     return self.tokenizer.batch_decode(*__snake_case ,  **__snake_case					)






  def     lowercase		(    self			:		Optional[int] ,  *lowerCAmelCase_			:		List[Any] ,  **lowerCAmelCase_			:		List[Any]					)							->						Any:
     return self.tokenizer.decode(*__snake_case ,  **__snake_case					)






  @property
  def     lowercase		(    self			:		Any					)							->						Union[str, Any]:
     warnings.warn(
         '`feature_extractor_class` is deprecated and will be removed in v5. Use `image_processor_class` instead.' ,  __snake_case ,  )
     return self.image_processor_class






  @property
  def     lowercase		(    self			:		int					)							->						str:
     warnings.warn(
         '`feature_extractor` is deprecated and will be removed in v5. Use `image_processor` instead.' ,  __snake_case ,  )
     return self.image_processor








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     _UpperCAmelCase   (   _UpperCamelCase		,   unittest.TestCase   ):

  """simple docstring"""


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






  def     lowercase		(    self			:		List[Any]					)							->						Dict:
     torch.manual_seed(0					)
     __lowerCAmelCase							     =     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=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') ,  )
     __lowerCAmelCase							     =     IPNDMScheduler()

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






  def     lowercase		(    self			:		Tuple ,  lowerCAmelCase_			:		Tuple ,  lowerCAmelCase_			:		List[str]=0					)							->						Any:
     if str(lowerCAmelCase_					).startswith('mps'					):
        __lowerCAmelCase							     =     torch.manual_seed(lowerCAmelCase_					)
     else:
        __lowerCAmelCase							     =     torch.Generator(device=lowerCAmelCase_					).manual_seed(lowerCAmelCase_					)
     __lowerCAmelCase							     =     {
         'batch_size': 1,
         'generator': generator,
         'num_inference_steps': 4,
     }
     return inputs






  def     lowercase		(    self			:		Union[str, Any]					)							->						int:
     __lowerCAmelCase							     =     'cpu'  # ensure determinism for the device-dependent torch.Generator
     __lowerCAmelCase							     =     self.get_dummy_components()
     __lowerCAmelCase							     =     DanceDiffusionPipeline(**lowerCAmelCase_					)
     __lowerCAmelCase							     =     pipe.to(lowerCAmelCase_					)
     pipe.set_progress_bar_config(disable=lowerCAmelCase_					)

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

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

     assert audio.shape == (1, 2, components["unet"].sample_size)
     __lowerCAmelCase							     =     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			:		Union[str, Any]					)							->						Tuple:
     return super().test_save_load_local()






  @skip_mps
  def     lowercase		(    self			:		List[str]					)							->						Dict:
     return super().test_dict_tuple_outputs_equivalent(expected_max_difference=3e-3					)






  @skip_mps
  def     lowercase		(    self			:		str					)							->						List[str]:
     return super().test_save_load_optional_components()






  @skip_mps
  def     lowercase		(    self			:		List[Any]					)							->						List[str]:
     return super().test_attention_slicing_forward_pass()






  def     lowercase		(    self			:		str					)							->						int:
     super().test_inference_batch_single_identical(expected_max_diff=3e-3					)





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

  """simple docstring"""


  def     lowercase		(    self			:		Any					)							->						Union[str, Any]:
     # clean up the VRAM after each test
     super().tearDown()
     gc.collect()
     torch.cuda.empty_cache()






  def     lowercase		(    self			:		List[str]					)							->						List[str]:
     __lowerCAmelCase							     =     torch_device

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

     __lowerCAmelCase							     =     torch.manual_seed(0					)
     __lowerCAmelCase							     =     pipe(generator=lowerCAmelCase_ ,  num_inference_steps=1_0_0 ,  audio_length_in_s=4.0_96					)
     __lowerCAmelCase							     =     output.audios

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

     assert audio.shape == (1, 2, pipe.unet.sample_size)
     __lowerCAmelCase							     =     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			:		Tuple					)							->						Dict:
     __lowerCAmelCase							     =     torch_device

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

     __lowerCAmelCase							     =     torch.manual_seed(0					)
     __lowerCAmelCase							     =     pipe(generator=lowerCAmelCase_ ,  num_inference_steps=1_0_0 ,  audio_length_in_s=4.0_96					)
     __lowerCAmelCase							     =     output.audios

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

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

