
'''simple docstring'''
import warnings
from typing import List, Optional, Union

from ...processing_utils import ProcessorMixin
from ...tokenization_utils_base import BatchEncoding, PaddingStrategy, PreTokenizedInput, TextInput, TruncationStrategy
from ...utils import TensorType






class 				UpperCAmelCase_			(		lowerCAmelCase__   ):
 UpperCAmelCase_						      =  ['''image_processor''', '''tokenizer''']
 UpperCAmelCase_						      =  '''LayoutLMv2ImageProcessor'''
 UpperCAmelCase_						      =  ('''LayoutLMv2Tokenizer''', '''LayoutLMv2TokenizerFast''')





 def __init__(      self	,    lowercase_=None	,    lowercase_=None	,    **lowercase_):
     snake_case_	:      Tuple            =  None
     if "feature_extractor" in kwargs:
         warnings.warn(
             "The `feature_extractor` argument is deprecated and will be removed in v5, use `image_processor`"
             " instead."	,    __lowerCAmelCase	,    )
         snake_case_	:      Dict            =  kwargs.pop("feature_extractor")

     snake_case_	:      Optional[Any]            =  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__(__lowerCAmelCase	,    __lowerCAmelCase)





 def __call__(      self	,    lowercase_	,    lowercase_ = None	,    lowercase_ = None	,    lowercase_ = None	,    lowercase_ = None	,    lowercase_ = True	,    lowercase_ = False	,    lowercase_ = False	,    lowercase_ = None	,    lowercase_ = 0	,    lowercase_ = None	,    lowercase_ = None	,    lowercase_ = None	,    lowercase_ = False	,    lowercase_ = False	,    lowercase_ = False	,    lowercase_ = False	,    lowercase_ = True	,    lowercase_ = None	,    **lowercase_	,    ):


     if self.image_processor.apply_ocr and (boxes is not None):
         raise ValueError(
             "You cannot provide bounding boxes if you initialized the image processor with apply_ocr set to True.")

     if self.image_processor.apply_ocr and (word_labels is not None):
         raise ValueError(
             "You cannot provide word labels if you initialized the image processor with apply_ocr set to True.")

     if return_overflowing_tokens is True and return_offsets_mapping is False:
         raise ValueError("You cannot return overflowing tokens without returning the offsets mapping.")

     # first, apply the image processor
     snake_case_	:      int            =  self.image_processor(images=__lowerCAmelCase	,    return_tensors=__lowerCAmelCase)

     # second, apply the tokenizer
     if text is not None and self.image_processor.apply_ocr and text_pair is None:
         if isinstance(__lowerCAmelCase	,    __lowerCAmelCase):
             snake_case_	:      Union[str, Any]            =  [text]  # add batch dimension (as the image processor always adds a batch dimension)
         snake_case_	:      Any            =  features["words"]

     snake_case_	:      Any            =  self.tokenizer(
         text=text if text is not None else features["words"]	,    text_pair=text_pair if text_pair is not None else None	,    boxes=boxes if boxes is not None else features["boxes"]	,    word_labels=__lowerCAmelCase	,    add_special_tokens=__lowerCAmelCase	,    padding=__lowerCAmelCase	,    truncation=__lowerCAmelCase	,    max_length=__lowerCAmelCase	,    stride=__lowerCAmelCase	,    pad_to_multiple_of=__lowerCAmelCase	,    return_token_type_ids=__lowerCAmelCase	,    return_attention_mask=__lowerCAmelCase	,    return_overflowing_tokens=__lowerCAmelCase	,    return_special_tokens_mask=__lowerCAmelCase	,    return_offsets_mapping=__lowerCAmelCase	,    return_length=__lowerCAmelCase	,    verbose=__lowerCAmelCase	,    return_tensors=__lowerCAmelCase	,    **__lowerCAmelCase	,    )

     # add pixel values
     snake_case_	:      Any            =  features.pop("pixel_values")
     if return_overflowing_tokens is True:
         snake_case_	:      Union[str, Any]            =  self.get_overflowing_images(__lowerCAmelCase	,    encoded_inputs["overflow_to_sample_mapping"])
     snake_case_	:      int            =  images

     return encoded_inputs





 def     snake_case__						(      self	,    lowercase_	,    lowercase_):
     snake_case_	:      Dict            =  []
     for sample_idx in overflow_to_sample_mapping:
         images_with_overflow.append(images[sample_idx])

     if len(__lowerCAmelCase) != len(__lowerCAmelCase):
         raise ValueError(
             "Expected length of images to be the same as the length of `overflow_to_sample_mapping`, but got"
             F' {len(__lowerCAmelCase)} and {len(__lowerCAmelCase)}')

     return images_with_overflow





 def     snake_case__						(      self	,    *lowercase_	,    **lowercase_):
     return self.tokenizer.batch_decode(*__lowerCAmelCase	,    **__lowerCAmelCase)





 def     snake_case__						(      self	,    *lowercase_	,    **lowercase_):
     return self.tokenizer.decode(*__lowerCAmelCase	,    **__lowerCAmelCase)





 @property
 def     snake_case__						(      self):
     return ["input_ids", "bbox", "token_type_ids", "attention_mask", "image"]





 @property
 def     snake_case__						(      self):
     warnings.warn(
         "`feature_extractor_class` is deprecated and will be removed in v5. Use `image_processor_class` instead."	,    __lowerCAmelCase	,    )
     return self.image_processor_class





 @property
 def     snake_case__						(      self):
     warnings.warn(
         "`feature_extractor` is deprecated and will be removed in v5. Use `image_processor` instead."	,    __lowerCAmelCase	,    )
     return self.image_processor






'''simple docstring'''
import gc
import unittest

import numpy as np
import torch

from diffusers import DDPMScheduler, MidiProcessor, SpectrogramDiffusionPipeline
from diffusers.pipelines.spectrogram_diffusion import SpectrogramContEncoder, SpectrogramNotesEncoder, TaFilmDecoder
from diffusers.utils import require_torch_gpu, skip_mps, slow, torch_device
from diffusers.utils.testing_utils import enable_full_determinism, require_note_seq, require_onnxruntime

from ..pipeline_params import TOKENS_TO_AUDIO_GENERATION_BATCH_PARAMS, TOKENS_TO_AUDIO_GENERATION_PARAMS
from ..test_pipelines_common import PipelineTesterMixin


enable_full_determinism()


a_     =  "./tests/fixtures/elise_format0.mid"






@unittest.skip("""The note-seq package currently throws an error on import"""   )
class 				UpperCAmelCase_			(		snake_case__      ,  unittest.TestCase   ):
 UpperCAmelCase_						      =  SpectrogramDiffusionPipeline
 UpperCAmelCase_						      =  PipelineTesterMixin.required_optional_params - {
     """callback""",
     """latents""",
     """callback_steps""",
     """output_type""",
     """num_images_per_prompt""",
 }
 UpperCAmelCase_						      =  False
 UpperCAmelCase_						      =  False
 UpperCAmelCase_						      =  TOKENS_TO_AUDIO_GENERATION_PARAMS
 UpperCAmelCase_						      =  TOKENS_TO_AUDIO_GENERATION_BATCH_PARAMS





 def     snake_case__						(      self):
     torch.manual_seed(0)
     snake_case_	:      List[str]            =  SpectrogramNotesEncoder(
         max_length=20_48	,    vocab_size=15_36	,    d_model=7_68	,    dropout_rate=0.1	,    num_layers=1	,    num_heads=1	,    d_kv=4	,    d_ff=20_48	,    feed_forward_proj="gated-gelu"	,    )

     snake_case_	:      int            =  SpectrogramContEncoder(
         input_dims=1_28	,    targets_context_length=2_56	,    d_model=7_68	,    dropout_rate=0.1	,    num_layers=1	,    num_heads=1	,    d_kv=4	,    d_ff=20_48	,    feed_forward_proj="gated-gelu"	,    )

     snake_case_	:      Tuple            =  TaFilmDecoder(
         input_dims=1_28	,    targets_length=2_56	,    max_decoder_noise_time=20_000.0	,    d_model=7_68	,    num_layers=1	,    num_heads=1	,    d_kv=4	,    d_ff=20_48	,    dropout_rate=0.1	,    )

     snake_case_	:      Union[str, Any]            =  DDPMScheduler()

     snake_case_	:      List[Any]            =  {
         "notes_encoder": notes_encoder.eval(),
         "continuous_encoder": continuous_encoder.eval(),
         "decoder": decoder.eval(),
         "scheduler": scheduler,
         "melgan": None,
     }
     return components





 def     snake_case__						(      self	,    lowercase_	,    lowercase_=0):
     if str(lowercase_).startswith("mps"):
         snake_case_	:      Dict            =  torch.manual_seed(lowercase_)
     else:
         snake_case_	:      Dict            =  torch.Generator(device=lowercase_).manual_seed(lowercase_)
     snake_case_	:      int            =  {
         "input_tokens": [
             [11_34, 90, 11_35, 11_33, 10_80, 1_12, 11_32, 10_80, 11_33, 10_79, 1_33, 11_32, 10_79, 11_33, 1] + [0] * 20_33
         ],
         "generator": generator,
         "num_inference_steps": 4,
         "output_type": "mel",
     }
     return inputs





 def     snake_case__						(      self):
     snake_case_	:      List[Any]            =  "cpu"  # ensure determinism for the device-dependent torch.Generator
     snake_case_	:      List[Any]            =  self.get_dummy_components()
     snake_case_	:      Tuple            =  SpectrogramDiffusionPipeline(**lowercase_)
     snake_case_	:      Tuple            =  pipe.to(lowercase_)
     pipe.set_progress_bar_config(disable=lowercase_)

     snake_case_	:      List[str]            =  self.get_dummy_inputs(lowercase_)
     snake_case_	:      Any            =  pipe(**lowercase_)
     snake_case_	:      str            =  output.audios

     snake_case_	:      str            =  mel[0, -3:, -3:]

     assert mel_slice.shape == (3, 3)
     snake_case_	:      Any            =  np.array(
         [-11.512_925, -4.788_215, -0.46_172_905, -2.051_715, -10.539_147, -10.970_963, -9.091_634, 4.0, 4.0])
     assert np.abs(mel_slice.flatten() - expected_slice).max() < 1E-2





 @skip_mps
 def     snake_case__						(      self):
     return super().test_save_load_local()





 @skip_mps
 def     snake_case__						(      self):
     return super().test_dict_tuple_outputs_equivalent()





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





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





 def     snake_case__						(      self):
     pass





 def     snake_case__						(      self):
     pass





 @skip_mps
 def     snake_case__						(      self):
     return super().test_progress_bar()






@slow
@require_torch_gpu
@require_onnxruntime
@require_note_seq
class 				UpperCAmelCase_			(		unittest.TestCase   ):





 def     snake_case__						(      self):
     # clean up the VRAM after each test
     super().tearDown()
     gc.collect()
     torch.cuda.empty_cache()





 def     snake_case__						(      self):
     # TODO - test that pipeline can decode tokens in a callback
     # so that music can be played live
     snake_case_	:      Any            =  torch_device

     snake_case_	:      str            =  SpectrogramDiffusionPipeline.from_pretrained("google/music-spectrogram-diffusion")
     snake_case_	:      Tuple            =  pipe.melgan
     snake_case_	:      List[str]            =  None

     snake_case_	:      int            =  pipe.to(lowercase_)
     pipe.set_progress_bar_config(disable=lowercase_)

     def callback(lowercase_	,    lowercase_):
         # decode mel to audio
         snake_case_	:      Dict            =  melgan(input_features=mel_output.astype(np.floataa))[0]
         assert len(audio[0]) == 8_19_20 * (step + 1)
         # simulate that audio is played
         return audio

     snake_case_	:      Tuple            =  MidiProcessor()
     snake_case_	:      int            =  processor(lowercase_)

     snake_case_	:      int            =  input_tokens[:3]
     snake_case_	:      Union[str, Any]            =  torch.manual_seed(0)
     pipe(lowercase_	,    num_inference_steps=5	,    generator=lowercase_	,    callback=lowercase_	,    output_type="mel")





 def     snake_case__						(      self):
     snake_case_	:      Any            =  torch_device

     snake_case_	:      Optional[Any]            =  SpectrogramDiffusionPipeline.from_pretrained("google/music-spectrogram-diffusion")
     snake_case_	:      Optional[int]            =  pipe.to(lowercase_)
     pipe.set_progress_bar_config(disable=lowercase_)
     snake_case_	:      Union[str, Any]            =  MidiProcessor()

     snake_case_	:      List[str]            =  processor(lowercase_)
     # just run two denoising loops
     snake_case_	:      int            =  input_tokens[:2]

     snake_case_	:      Optional[Any]            =  torch.manual_seed(0)
     snake_case_	:      Tuple            =  pipe(lowercase_	,    num_inference_steps=2	,    generator=lowercase_)

     snake_case_	:      Optional[Any]            =  output.audios[0]

     assert abs(np.abs(lowercase_).sum() - 3_612.841) < 1E-1





 def     snake_case__						(      self):
     snake_case_	:      Union[str, Any]            =  torch_device

     snake_case_	:      Tuple            =  SpectrogramDiffusionPipeline.from_pretrained("google/music-spectrogram-diffusion")
     snake_case_	:      Optional[Any]            =  pipe.to(lowercase_)
     pipe.set_progress_bar_config(disable=lowercase_)

     snake_case_	:      Tuple            =  MidiProcessor()

     snake_case_	:      int            =  processor(lowercase_)

     # just run 4 denoising loops
     snake_case_	:      Any            =  input_tokens[:4]

     snake_case_	:      Optional[Any]            =  torch.manual_seed(0)
     snake_case_	:      Tuple            =  pipe(lowercase_	,    num_inference_steps=1_00	,    generator=lowercase_)

     snake_case_	:      Tuple            =  output.audios[0]
     assert abs(np.abs(lowercase_).sum() - 9_389.1_111) < 5E-2




