




import csv
import dataclasses
import json
from dataclasses import dataclass
from typing import List, Optional, Union

from ...utils import is_tf_available, is_torch_available, logging


lowercase_		=							logging.get_logger(__name__)





@dataclass
class 		A__    :
	lowerCamelCase__		:			str
	lowerCamelCase__		:			str
	lowerCamelCase__		:			Optional[str]       =None
	lowerCamelCase__		:			Optional[str]       =None






	def 			lowercase				(     self						)				->       Dict:




						"""simple docstring"""






						return json.dumps(dataclasses.asdict(self						)    ,					indent=2						) + "\n"





@dataclass(frozen=__SCREAMING_SNAKE_CASE			)
class 		A__    :
	lowerCamelCase__		:			List[int]
	lowerCamelCase__		:			Optional[List[int]]       =None
	lowerCamelCase__		:			Optional[List[int]]       =None
	lowerCamelCase__		:			Optional[Union[int, float]]       =None






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




						"""simple docstring"""






						return json.dumps(dataclasses.asdict(self						)						) + "\n"





class 		A__    :






	def 			lowercase				(     self    ,					lowerCamelCase						)				->       int:




						"""simple docstring"""






						raise NotImplementedError()






	def 			lowercase				(     self    ,					lowerCamelCase						)				->       List[str]:




						"""simple docstring"""






						raise NotImplementedError()






	def 			lowercase				(     self    ,					lowerCamelCase						)				->       int:




						"""simple docstring"""






						raise NotImplementedError()






	def 			lowercase				(     self    ,					lowerCamelCase						)				->       List[str]:




						"""simple docstring"""






						raise NotImplementedError()






	def 			lowercase				(     self						)				->       Optional[int]:




						"""simple docstring"""






						raise NotImplementedError()






	def 			lowercase				(     self    ,					lowerCamelCase						)				->       List[str]:




						"""simple docstring"""






						if len(self.get_labels()						) > 1:
											__magic_name__ :							List[Any]		       =							self.get_labels()[int(example.label						)]
						return example






	@classmethod
	def 			lowercase				(     cls    ,					lowerCamelCase    ,					lowerCamelCase=None						)				->       List[Any]:




						"""simple docstring"""






						with open(lowerCamelCase    ,					'''r'''    ,					encoding='''utf-8-sig'''						) as f:
											return list(csv.reader(lowerCamelCase    ,					delimiter='''\t'''    ,					quotechar=lowerCamelCase						)						)





class 		A__    (       __SCREAMING_SNAKE_CASE			):






	def __init__(     self    ,					lowerCamelCase=None    ,					lowerCamelCase=None    ,					lowerCamelCase="classification"    ,					lowerCamelCase=False						)				->       int:




						"""simple docstring"""






						__magic_name__ :							Any		       =							[] if labels is None else labels
						__magic_name__ :							Union[str, Any]		       =							[] if examples is None else examples
						__magic_name__ :							str		       =							mode
						__magic_name__ :							Any		       =							verbose






	def __len__(     self						)				->       Tuple:




						"""simple docstring"""






						return len(self.examples						)






	def __getitem__(     self    ,					lowerCamelCase						)				->       Optional[int]:




						"""simple docstring"""






						if isinstance(lowerCamelCase    ,					lowerCamelCase						):
											return SingleSentenceClassificationProcessor(labels=self.labels    ,					examples=self.examples[idx]						)
						return self.examples[idx]






	@classmethod
	def 			lowercase				(     cls    ,					lowerCamelCase    ,					lowerCamelCase=""    ,					lowerCamelCase=0    ,					lowerCamelCase=1    ,					lowerCamelCase=None    ,					lowerCamelCase=False    ,					**lowerCamelCase						)				->       Union[str, Any]:




						"""simple docstring"""






						__magic_name__ :							Union[str, Any]		       =							cls(**lowerCamelCase						)
						processor.add_examples_from_csv(
						    lowerCamelCase    ,					split_name=lowerCamelCase    ,					column_label=lowerCamelCase    ,					column_text=lowerCamelCase    ,					column_id=lowerCamelCase    ,					skip_first_row=lowerCamelCase    ,					overwrite_labels=lowerCamelCase    ,					overwrite_examples=lowerCamelCase    ,					)
						return processor






	@classmethod
	def 			lowercase				(     cls    ,					lowerCamelCase    ,					lowerCamelCase=None    ,					**lowerCamelCase						)				->       str:




						"""simple docstring"""






						__magic_name__ :							List[Any]		       =							cls(**lowerCamelCase						)
						processor.add_examples(lowerCamelCase    ,					labels=lowerCamelCase						)
						return processor






	def 			lowercase				(     self    ,					lowerCamelCase    ,					lowerCamelCase=""    ,					lowerCamelCase=0    ,					lowerCamelCase=1    ,					lowerCamelCase=None    ,					lowerCamelCase=False    ,					lowerCamelCase=False    ,					lowerCamelCase=False    ,					)				->       Dict:




						"""simple docstring"""






						__magic_name__ :							Tuple		       =							self._read_tsv(lowerCamelCase						)
						if skip_first_row:
											__magic_name__ :							List[str]		       =							lines[1:]
						__magic_name__ :							List[Any]		       =							[]
						__magic_name__ :							List[str]		       =							[]
						__magic_name__ :							Optional[Any]		       =							[]
						for i, line in enumerate(lowerCamelCase						):
											texts.append(line[column_text]						)
											labels.append(line[column_label]						)
											if column_id is not None:
																ids.append(line[column_id]						)
											else:
																__magic_name__ :							Tuple		       =							F'''{split_name}-{i}''' if split_name else str(lowerCamelCase						)
																ids.append(lowerCamelCase						)

						return self.add_examples(
						    lowerCamelCase    ,					lowerCamelCase    ,					lowerCamelCase    ,					overwrite_labels=lowerCamelCase    ,					overwrite_examples=lowerCamelCase						)






	def 			lowercase				(     self    ,					lowerCamelCase    ,					lowerCamelCase=None    ,					lowerCamelCase=None    ,					lowerCamelCase=False    ,					lowerCamelCase=False						)				->       int:




						"""simple docstring"""






						if labels is not None and len(lowerCamelCase						) != len(lowerCamelCase						):
											raise ValueError(
											    F'''Text and labels have mismatched lengths {len(lowerCamelCase						)} and {len(lowerCamelCase						)}'''						)
						if ids is not None and len(lowerCamelCase						) != len(lowerCamelCase						):
											raise ValueError(F'''Text and ids have mismatched lengths {len(lowerCamelCase						)} and {len(lowerCamelCase						)}'''						)
						if ids is None:
											__magic_name__ :							Any		       =							[None] * len(lowerCamelCase						)
						if labels is None:
											__magic_name__ :							str		       =							[None] * len(lowerCamelCase						)
						__magic_name__ :							Union[str, Any]		       =							[]
						__magic_name__ :							Any		       =							set()
						for text_or_text_and_label, label, guid in zip(lowerCamelCase    ,					lowerCamelCase    ,					lowerCamelCase						):
											if isinstance(lowerCamelCase    ,					(tuple, list)						) and label is None:
																__magic_name__       ,			__magic_name__ :							List[str]		       =							text_or_text_and_label
											else:
																__magic_name__ :							Any		       =							text_or_text_and_label
											added_labels.add(lowerCamelCase						)
											examples.append(InputExample(guid=lowerCamelCase    ,					text_a=lowerCamelCase    ,					text_b=lowerCamelCase    ,					label=lowerCamelCase						)						)

						# Update examples
						if overwrite_examples:
											__magic_name__ :							Union[str, Any]		       =							examples
						else:
											self.examples.extend(lowerCamelCase						)

						# Update labels
						if overwrite_labels:
											__magic_name__ :							Tuple		       =							list(lowerCamelCase						)
						else:
											__magic_name__ :							Tuple		       =							list(set(self.labels						).union(lowerCamelCase						)						)

						return self.examples






	def 			lowercase				(     self    ,					lowerCamelCase    ,					lowerCamelCase=None    ,					lowerCamelCase=False    ,					lowerCamelCase=0    ,					lowerCamelCase=True    ,					lowerCamelCase=None    ,					)				->       Any:




						"""simple docstring"""






						if max_length is None:
											__magic_name__ :							Optional[int]		       =							tokenizer.max_len

						__magic_name__ :							Any		       =							{label: i for i, label in enumerate(self.labels						)}

						__magic_name__ :							List[str]		       =							[]
						for ex_index, example in enumerate(self.examples						):
											if ex_index % 10000 == 0:
																logger.info(F'''Tokenizing example {ex_index}'''						)

											__magic_name__ :							int		       =							tokenizer.encode(
											    example.text_a    ,					add_special_tokens=lowerCamelCase    ,					max_length=min(lowerCamelCase    ,					tokenizer.max_len						)    ,					)
											all_input_ids.append(lowerCamelCase						)

						__magic_name__ :							Union[str, Any]		       =							max(len(lowerCamelCase						) for input_ids in all_input_ids						)

						__magic_name__ :							Optional[Any]		       =							[]
						for ex_index, (input_ids, example) in enumerate(zip(lowerCamelCase    ,					self.examples						)						):
											if ex_index % 10000 == 0:
																logger.info(F'''Writing example {ex_index}/{len(self.examples						)}'''						)
											# The mask has 1 for real tokens and 0 for padding tokens. Only real
											# tokens are attended to.
											__magic_name__ :							str		       =							[1 if mask_padding_with_zero else 0] * len(lowerCamelCase						)

											# Zero-pad up to the sequence length.
											__magic_name__ :							Optional[Any]		       =							batch_length - len(lowerCamelCase						)
											if pad_on_left:
																__magic_name__ :							List[Any]		       =							([pad_token] * padding_length) + input_ids
																__magic_name__ :							Any		       =							([0 if mask_padding_with_zero else 1] * padding_length) + attention_mask
											else:
																__magic_name__ :							Optional[int]		       =							input_ids + ([pad_token] * padding_length)
																__magic_name__ :							List[str]		       =							attention_mask + ([0 if mask_padding_with_zero else 1] * padding_length)

											if len(lowerCamelCase						) != batch_length:
																raise ValueError(F'''Error with input length {len(lowerCamelCase						)} vs {batch_length}'''						)
											if len(lowerCamelCase						) != batch_length:
																raise ValueError(F'''Error with input length {len(lowerCamelCase						)} vs {batch_length}'''						)

											if self.mode == "classification":
																__magic_name__ :							List[str]		       =							label_map[example.label]
											elif self.mode == "regression":
																__magic_name__ :							Tuple		       =							float(example.label						)
											else:
																raise ValueError(self.mode						)

											if ex_index < 5 and self.verbose:
																logger.info('''*** Example ***'''						)
																logger.info(F'''guid: {example.guid}'''						)
																logger.info(F'''input_ids: {" ".join([str(lowerCamelCase						) for x in input_ids]						)}'''						)
																logger.info(F'''attention_mask: {" ".join([str(lowerCamelCase						) for x in attention_mask]						)}'''						)
																logger.info(F'''label: {example.label} (id = {label})'''						)

											features.append(InputFeatures(input_ids=lowerCamelCase    ,					attention_mask=lowerCamelCase    ,					label=lowerCamelCase						)						)

						if return_tensors is None:
											return features
						elif return_tensors == "tf":
											if not is_tf_available():
																raise RuntimeError('''return_tensors set to \'tf\' but TensorFlow 2.0 can\'t be imported'''						)
											import tensorflow as tf

											def gen():
																for ex in features:
																					yield ({"input_ids": ex.input_ids, "attention_mask": ex.attention_mask}, ex.label)

											__magic_name__ :							List[str]		       =							tf.data.Dataset.from_generator(
											    lowerCamelCase    ,					({'''input_ids''': tf.intaa, '''attention_mask''': tf.intaa}, tf.intaa)    ,					({'''input_ids''': tf.TensorShape([None]						), '''attention_mask''': tf.TensorShape([None]						)}, tf.TensorShape([]						))    ,					)
											return dataset
						elif return_tensors == "pt":
											if not is_torch_available():
																raise RuntimeError('''return_tensors set to \'pt\' but PyTorch can\'t be imported'''						)
											import torch
											from torch.utils.data import TensorDataset

											__magic_name__ :							Optional[int]		       =							torch.tensor([f.input_ids for f in features]    ,					dtype=torch.long						)
											__magic_name__ :							Dict		       =							torch.tensor([f.attention_mask for f in features]    ,					dtype=torch.long						)
											if self.mode == "classification":
																__magic_name__ :							Optional[int]		       =							torch.tensor([f.label for f in features]    ,					dtype=torch.long						)
											elif self.mode == "regression":
																__magic_name__ :							Dict		       =							torch.tensor([f.label for f in features]    ,					dtype=torch.float						)

											__magic_name__ :							Tuple		       =							TensorDataset(lowerCamelCase    ,					lowerCamelCase    ,					lowerCamelCase						)
											return dataset
						else:
											raise ValueError('''return_tensors should be one of \'tf\' or \'pt\''''						)









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


lowercase_		=							'''./tests/fixtures/elise_format0.mid'''





@unittest.skip("The note-seq package currently throws an error on import"			)
class 		A__    (       __SCREAMING_SNAKE_CASE		,					unittest.TestCase			):
	lowerCamelCase__		:			Optional[int]       =SpectrogramDiffusionPipeline
	lowerCamelCase__		:			List[str]       =PipelineTesterMixin.required_optional_params - {
	    "callback",
	    "latents",
	    "callback_steps",
	    "output_type",
	    "num_images_per_prompt",
	}
	lowerCamelCase__		:			Union[str, Any]       =False
	lowerCamelCase__		:			List[Any]       =False
	lowerCamelCase__		:			List[str]       =TOKENS_TO_AUDIO_GENERATION_PARAMS
	lowerCamelCase__		:			str       =TOKENS_TO_AUDIO_GENERATION_BATCH_PARAMS






	def 			lowercase				(     self						)				->       Any:




						"""simple docstring"""






						torch.manual_seed(0						)
						__magic_name__ :							Optional[Any]		       =							SpectrogramNotesEncoder(
						    max_length=2048    ,					vocab_size=1536    ,					d_model=768    ,					dropout_rate=0.1    ,					num_layers=1    ,					num_heads=1    ,					d_kv=4    ,					d_ff=2048    ,					feed_forward_proj='''gated-gelu'''    ,					)

						__magic_name__ :							Any		       =							SpectrogramContEncoder(
						    input_dims=128    ,					targets_context_length=256    ,					d_model=768    ,					dropout_rate=0.1    ,					num_layers=1    ,					num_heads=1    ,					d_kv=4    ,					d_ff=2048    ,					feed_forward_proj='''gated-gelu'''    ,					)

						__magic_name__ :							List[Any]		       =							TaFilmDecoder(
						    input_dims=128    ,					targets_length=256    ,					max_decoder_noise_time=2_0_0_0_0.0    ,					d_model=768    ,					num_layers=1    ,					num_heads=1    ,					d_kv=4    ,					d_ff=2048    ,					dropout_rate=0.1    ,					)

						__magic_name__ :							Any		       =							DDPMScheduler()

						__magic_name__ :							Dict		       =							{
						    '''notes_encoder''': notes_encoder.eval(),
						    '''continuous_encoder''': continuous_encoder.eval(),
						    '''decoder''': decoder.eval(),
						    '''scheduler''': scheduler,
						    '''melgan''': None,
						}
						return components






	def 			lowercase				(     self    ,					lowerCamelCase    ,					lowerCamelCase=0						)				->       Dict:




						"""simple docstring"""






						if str(lowerCamelCase						).startswith('''mps'''						):
											__magic_name__ :							List[str]		       =							torch.manual_seed(lowerCamelCase						)
						else:
											__magic_name__ :							Any		       =							torch.Generator(device=lowerCamelCase						).manual_seed(lowerCamelCase						)
						__magic_name__ :							Any		       =							{
						    '''input_tokens''': [
						        [1134, 90, 1135, 1133, 1080, 112, 1132, 1080, 1133, 1079, 133, 1132, 1079, 1133, 1] + [0] * 2033
						    ],
						    '''generator''': generator,
						    '''num_inference_steps''': 4,
						    '''output_type''': '''mel''',
						}
						return inputs






	def 			lowercase				(     self						)				->       Optional[int]:




						"""simple docstring"""






						__magic_name__ :							Optional[Any]		       =							'''cpu'''  # ensure determinism for the device-dependent torch.Generator
						__magic_name__ :							Union[str, Any]		       =							self.get_dummy_components()
						__magic_name__ :							str		       =							SpectrogramDiffusionPipeline(**lowerCamelCase						)
						__magic_name__ :							List[Any]		       =							pipe.to(lowerCamelCase						)
						pipe.set_progress_bar_config(disable=lowerCamelCase						)

						__magic_name__ :							Tuple		       =							self.get_dummy_inputs(lowerCamelCase						)
						__magic_name__ :							List[str]		       =							pipe(**lowerCamelCase						)
						__magic_name__ :							Any		       =							output.audios

						__magic_name__ :							Union[str, Any]		       =							mel[0, -3:, -3:]

						assert mel_slice.shape == (3, 3)
						__magic_name__ :							Dict		       =							np.array(
						    [-1_1.5_1_2_9_2_5, -4.7_8_8_2_1_5, -0.4_6_1_7_2_9_0_5, -2.0_5_1_7_1_5, -1_0.5_3_9_1_4_7, -1_0.9_7_0_9_6_3, -9.0_9_1_6_3_4, 4.0, 4.0]						)
						assert np.abs(mel_slice.flatten() - expected_slice						).max() < 1e-2






	@skip_mps
	def 			lowercase				(     self						)				->       List[Any]:




						"""simple docstring"""






						return super().test_save_load_local()






	@skip_mps
	def 			lowercase				(     self						)				->       str:




						"""simple docstring"""






						return super().test_dict_tuple_outputs_equivalent()






	@skip_mps
	def 			lowercase				(     self						)				->       List[Any]:




						"""simple docstring"""






						return super().test_save_load_optional_components()






	@skip_mps
	def 			lowercase				(     self						)				->       Union[str, Any]:




						"""simple docstring"""






						return super().test_attention_slicing_forward_pass()






	def 			lowercase				(     self						)				->       Tuple:




						"""simple docstring"""






						pass






	def 			lowercase				(     self						)				->       int:




						"""simple docstring"""






						pass






	@skip_mps
	def 			lowercase				(     self						)				->       Optional[Any]:




						"""simple docstring"""






						return super().test_progress_bar()





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






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




						"""simple docstring"""






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






	def 			lowercase				(     self						)				->       Optional[Any]:




						"""simple docstring"""






						__magic_name__ :							List[Any]		       =							torch_device

						__magic_name__ :							Any		       =							SpectrogramDiffusionPipeline.from_pretrained('''google/music-spectrogram-diffusion'''						)
						__magic_name__ :							Optional[int]		       =							pipe.melgan
						__magic_name__ :							List[str]		       =							None

						__magic_name__ :							Tuple		       =							pipe.to(lowerCamelCase						)
						pipe.set_progress_bar_config(disable=lowerCamelCase						)

						def callback(lowerCamelCase    ,					lowerCamelCase						):
											# decode mel to audio
											__magic_name__ :							Tuple		       =							melgan(input_features=mel_output.astype(np.floataa						)						)[0]
											assert len(audio[0]						) == 81920 * (step + 1)
											# simulate that audio is played
											return audio

						__magic_name__ :							int		       =							MidiProcessor()
						__magic_name__ :							List[str]		       =							processor(lowerCamelCase						)

						__magic_name__ :							Dict		       =							input_tokens[:3]
						__magic_name__ :							Any		       =							torch.manual_seed(0						)
						pipe(lowerCamelCase    ,					num_inference_steps=5    ,					generator=lowerCamelCase    ,					callback=lowerCamelCase    ,					output_type='''mel'''						)






	def 			lowercase				(     self						)				->       int:




						"""simple docstring"""






						__magic_name__ :							Optional[int]		       =							torch_device

						__magic_name__ :							int		       =							SpectrogramDiffusionPipeline.from_pretrained('''google/music-spectrogram-diffusion'''						)
						__magic_name__ :							str		       =							pipe.to(lowerCamelCase						)
						pipe.set_progress_bar_config(disable=lowerCamelCase						)
						__magic_name__ :							Any		       =							MidiProcessor()

						__magic_name__ :							Optional[Any]		       =							processor(lowerCamelCase						)
						# just run two denoising loops
						__magic_name__ :							Any		       =							input_tokens[:2]

						__magic_name__ :							int		       =							torch.manual_seed(0						)
						__magic_name__ :							Any		       =							pipe(lowerCamelCase    ,					num_inference_steps=2    ,					generator=lowerCamelCase						)

						__magic_name__ :							Union[str, Any]		       =							output.audios[0]

						assert abs(np.abs(lowerCamelCase						).sum() - 3_6_1_2.8_4_1						) < 1e-1






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




						"""simple docstring"""






						__magic_name__ :							Optional[int]		       =							torch_device

						__magic_name__ :							str		       =							SpectrogramDiffusionPipeline.from_pretrained('''google/music-spectrogram-diffusion'''						)
						__magic_name__ :							List[Any]		       =							pipe.to(lowerCamelCase						)
						pipe.set_progress_bar_config(disable=lowerCamelCase						)

						__magic_name__ :							Optional[int]		       =							MidiProcessor()

						__magic_name__ :							str		       =							processor(lowerCamelCase						)

						# just run 4 denoising loops
						__magic_name__ :							List[str]		       =							input_tokens[:4]

						__magic_name__ :							Union[str, Any]		       =							torch.manual_seed(0						)
						__magic_name__ :							Dict		       =							pipe(lowerCamelCase    ,					num_inference_steps=100    ,					generator=lowerCamelCase						)

						__magic_name__ :							Optional[int]		       =							output.audios[0]
						assert abs(np.abs(lowerCamelCase						).sum() - 9_3_8_9.1_1_1_1						) < 5e-2



