code: 
  
  
  
  
  
  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		:     Any 					=logging.get_logger(__name__)
  
  
  
  
  
  
  @dataclass
  class 	UpperCamelCase_     :
  					_a			:	str
  					_a			:	str
  					_a			:	Optional[str]							=				None
  					_a			:	Optional[str]							=				None
  
  
  
  					def      __a    (		self    :    str	):
  							return json.dumps(dataclasses.asdict(self	)					,			indent=2	) + "\n"
  
  
  
  
  
  
  @dataclass(frozen=snake_case__       )
  class 	UpperCamelCase_     :
  					_a			:	List[int]
  					_a			:	Optional[List[int]]							=				None
  					_a			:	Optional[List[int]]							=				None
  					_a			:	Optional[Union[int, float]]							=				None
  
  
  
  					def      __a    (		self    :    List[Any]	):
  							return json.dumps(dataclasses.asdict(self	)	) + "\n"
  
  
  
  
  
  
  class 	UpperCamelCase_     :
  
  
  
  					def      __a    (		self    :    List[str]					,			lowerCamelCase    :    Tuple	):
  							raise NotImplementedError()
  
  
  
  					def      __a    (		self    :    Union[str, Any]					,			lowerCamelCase    :    int	):
  							raise NotImplementedError()
  
  
  
  					def      __a    (		self    :    str					,			lowerCamelCase    :    Optional[Any]	):
  							raise NotImplementedError()
  
  
  
  					def      __a    (		self    :    Union[str, Any]					,			lowerCamelCase    :    Optional[int]	):
  							raise NotImplementedError()
  
  
  
  					def      __a    (		self    :    Union[str, Any]	):
  							raise NotImplementedError()
  
  
  
  					def      __a    (		self    :    List[str]					,			lowerCamelCase    :    Any	):
  							if len(self.get_labels()	) > 1:
  									lowerCamelCase_  :    Optional[Any]   	=			self.get_labels()[int(example.label	)]
  							return example
  
  
  
  
  
  					@classmethod
  					def      __a    (		cls    :    List[str]					,			lowerCamelCase    :    Any					,			lowerCamelCase    :    List[str]=None	):
  							with open(lowerCamelCase					,			'r'					,			encoding='utf-8-sig'	) as f:
  									return list(csv.reader(lowerCamelCase					,			delimiter='\t'					,			quotechar=lowerCamelCase	)	)
  
  
  
  
  
  
  class 	UpperCamelCase_     (     snake_case__       ):
  
  
  
  					def __init__(		self    :    Any					,			lowerCamelCase    :    Dict=None					,			lowerCamelCase    :    List[Any]=None					,			lowerCamelCase    :    Optional[int]="classification"					,			lowerCamelCase    :    str=False	):
  							lowerCamelCase_  :    Any   	=			[] if labels is None else labels
  							lowerCamelCase_  :    str   	=			[] if examples is None else examples
  							lowerCamelCase_  :    Optional[Any]   	=			mode
  							lowerCamelCase_  :    Tuple   	=			verbose
  
  
  
  					def __len__(		self    :    Any	):
  							return len(self.examples	)
  
  
  
  					def __getitem__(		self    :    Any					,			lowerCamelCase    :    Optional[Any]	):
  							if isinstance(lowerCamelCase					,			lowerCamelCase	):
  									return SingleSentenceClassificationProcessor(labels=self.labels					,			examples=self.examples[idx]	)
  							return self.examples[idx]
  
  
  
  					@classmethod
  					def      __a    (		cls    :    Union[str, Any]					,			lowerCamelCase    :    List[Any]					,			lowerCamelCase    :    str=""					,			lowerCamelCase    :    Optional[int]=0					,			lowerCamelCase    :    List[str]=1					,			lowerCamelCase    :    Dict=None					,			lowerCamelCase    :    Dict=False					,			**lowerCamelCase    :    int	):
  							lowerCamelCase_  :    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      __a    (		cls    :    Tuple					,			lowerCamelCase    :    Optional[Any]					,			lowerCamelCase    :    Any=None					,			**lowerCamelCase    :    Optional[int]	):
  							lowerCamelCase_  :    Tuple   	=			cls(**lowerCamelCase	)
  							processor.add_examples(lowerCamelCase					,			labels=lowerCamelCase	)
  							return processor
  
  
  
  					def      __a    (		self    :    List[str]					,			lowerCamelCase    :    str					,			lowerCamelCase    :    int=""					,			lowerCamelCase    :    Optional[int]=0					,			lowerCamelCase    :    Dict=1					,			lowerCamelCase    :    Dict=None					,			lowerCamelCase    :    Optional[Any]=False					,			lowerCamelCase    :    Tuple=False					,			lowerCamelCase    :    Optional[int]=False					,			):
  							lowerCamelCase_  :    Tuple   	=			self._read_tsv(lowerCamelCase	)
  							if skip_first_row:
  									lowerCamelCase_  :    List[Any]   	=			lines[1:]
  							lowerCamelCase_  :    Any   	=			[]
  							lowerCamelCase_  :    str   	=			[]
  							lowerCamelCase_  :    str   	=			[]
  							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:
  											lowerCamelCase_  :    List[str]   	=			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      __a    (		self    :    Dict					,			lowerCamelCase    :    Optional[int]					,			lowerCamelCase    :    Union[str, Any]=None					,			lowerCamelCase    :    Optional[int]=None					,			lowerCamelCase    :    str=False					,			lowerCamelCase    :    str=False	):
  							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:
  									lowerCamelCase_  :    Any   	=			[None] * len(lowerCamelCase	)
  							if labels is None:
  									lowerCamelCase_  :    Any   	=			[None] * len(lowerCamelCase	)
  							lowerCamelCase_  :    str   	=			[]
  							lowerCamelCase_  :    int   	=			set()
  							for text_or_text_and_label, label, guid in zip(lowerCamelCase					,			lowerCamelCase					,			lowerCamelCase	):
  									if isinstance(lowerCamelCase					,			(tuple, list)	) and label is None:
  											lowerCamelCase_	,			lowerCamelCase_  :    Dict   	=			text_or_text_and_label
  									else:
  											lowerCamelCase_  :    str   	=			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:
  									lowerCamelCase_  :    Tuple   	=			examples
  							else:
  									self.examples.extend(lowerCamelCase	)
  
  							# Update labels
  							if overwrite_labels:
  									lowerCamelCase_  :    int   	=			list(lowerCamelCase	)
  							else:
  									lowerCamelCase_  :    List[str]   	=			list(set(self.labels	).union(lowerCamelCase	)	)
  
  							return self.examples
  
  
  
  
  
  					def      __a    (		self    :    Optional[int]					,			lowerCamelCase    :    Optional[Any]					,			lowerCamelCase    :    Tuple=None					,			lowerCamelCase    :    Tuple=False					,			lowerCamelCase    :    Optional[int]=0					,			lowerCamelCase    :    Optional[Any]=True					,			lowerCamelCase    :    List[str]=None					,			):
  							if max_length is None:
  									lowerCamelCase_  :    List[Any]   	=			tokenizer.max_len
  
  							lowerCamelCase_  :    List[Any]   	=			{label: i for i, label in enumerate(self.labels	)}
  
  							lowerCamelCase_  :    Union[str, Any]   	=			[]
  							for ex_index, example in enumerate(self.examples	):
  									if ex_index % 1_00_00 == 0:
  											logger.info(F"Tokenizing example {ex_index}"	)
  
  									lowerCamelCase_  :    Tuple   	=			tokenizer.encode(
  									    example.text_a					,			add_special_tokens=lowerCamelCase					,			max_length=min(lowerCamelCase					,			tokenizer.max_len	)					,			)
  									all_input_ids.append(lowerCamelCase	)
  
  							lowerCamelCase_  :    Optional[int]   	=			max(len(lowerCamelCase	) for input_ids in all_input_ids	)
  
  							lowerCamelCase_  :    Dict   	=			[]
  							for ex_index, (input_ids, example) in enumerate(zip(lowerCamelCase					,			self.examples	)	):
  									if ex_index % 1_00_00 == 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.
  									lowerCamelCase_  :    Tuple   	=			[1 if mask_padding_with_zero else 0] * len(lowerCamelCase	)
  
  									# Zero-pad up to the sequence length.
  									lowerCamelCase_  :    Any   	=			batch_length - len(lowerCamelCase	)
  									if pad_on_left:
  											lowerCamelCase_  :    Any   	=			([pad_token] * padding_length) + input_ids
  											lowerCamelCase_  :    int   	=			([0 if mask_padding_with_zero else 1] * padding_length) + attention_mask
  									else:
  											lowerCamelCase_  :    Any   	=			input_ids + ([pad_token] * padding_length)
  											lowerCamelCase_  :    Optional[Any]   	=			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":
  											lowerCamelCase_  :    Optional[int]   	=			label_map[example.label]
  									elif self.mode == "regression":
  											lowerCamelCase_  :    List[str]   	=			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)
  
  									lowerCamelCase_  :    Dict   	=			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
  
  									lowerCamelCase_  :    Optional[int]   	=			torch.tensor([f.input_ids for f in features]					,			dtype=torch.long	)
  									lowerCamelCase_  :    Dict   	=			torch.tensor([f.attention_mask for f in features]					,			dtype=torch.long	)
  									if self.mode == "classification":
  											lowerCamelCase_  :    Tuple   	=			torch.tensor([f.label for f in features]					,			dtype=torch.long	)
  									elif self.mode == "regression":
  											lowerCamelCase_  :    Union[str, Any]   	=			torch.tensor([f.label for f in features]					,			dtype=torch.float	)
  
  									lowerCamelCase_  :    Dict   	=			TensorDataset(lowerCamelCase					,			lowerCamelCase					,			lowerCamelCase	)
  									return dataset
  							else:
  									raise ValueError('return_tensors should be one of \'tf\' or \'pt\''	)
  
style_context: 
  
  
  
  
  
  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		:     Optional[Any] 					="""./tests/fixtures/elise_format0.mid"""
  
  
  
  
  
  
  @unittest.skip('The note-seq package currently throws an error on import'       )
  class 	UpperCamelCase_     (     snake_case__     ,    unittest.TestCase       ):
  					_a			:	int							=				SpectrogramDiffusionPipeline
  					_a			:	Dict							=				PipelineTesterMixin.required_optional_params - {
  					    'callback',
  					    'latents',
  					    'callback_steps',
  					    'output_type',
  					    'num_images_per_prompt',
  					}
  					_a			:	str							=				False
  					_a			:	Union[str, Any]							=				False
  					_a			:	Optional[int]							=				TOKENS_TO_AUDIO_GENERATION_PARAMS
  					_a			:	Optional[Any]							=				TOKENS_TO_AUDIO_GENERATION_BATCH_PARAMS
  
  
  
  					def      __a    (		self    :    Dict	):
  							torch.manual_seed(0	)
  							lowerCamelCase_  :    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'					,			)
  
  							lowerCamelCase_  :    Union[str, Any]   	=			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'					,			)
  
  							lowerCamelCase_  :    Union[str, Any]   	=			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					,			)
  
  							lowerCamelCase_  :    Any   	=			DDPMScheduler()
  
  							lowerCamelCase_  :    str   	=			{
  							    'notes_encoder': notes_encoder.eval(),
  							    'continuous_encoder': continuous_encoder.eval(),
  							    'decoder': decoder.eval(),
  							    'scheduler': scheduler,
  							    'melgan': None,
  							}
  							return components
  
  
  
  					def      __a    (		self    :    Any					,			lowerCamelCase    :    Any					,			lowerCamelCase    :    int=0	):
  							if str(lowerCamelCase	).startswith('mps'	):
  									lowerCamelCase_  :    Optional[int]   	=			torch.manual_seed(lowerCamelCase	)
  							else:
  									lowerCamelCase_  :    Dict   	=			torch.Generator(device=lowerCamelCase	).manual_seed(lowerCamelCase	)
  							lowerCamelCase_  :    str   	=			{
  							    '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      __a    (		self    :    Dict	):
  							lowerCamelCase_  :    str   	=			'cpu'  # ensure determinism for the device-dependent torch.Generator
  							lowerCamelCase_  :    Dict   	=			self.get_dummy_components()
  							lowerCamelCase_  :    Tuple   	=			SpectrogramDiffusionPipeline(**lowerCamelCase	)
  							lowerCamelCase_  :    List[str]   	=			pipe.to(lowerCamelCase	)
  							pipe.set_progress_bar_config(disable=lowerCamelCase	)
  
  							lowerCamelCase_  :    Optional[Any]   	=			self.get_dummy_inputs(lowerCamelCase	)
  							lowerCamelCase_  :    str   	=			pipe(**lowerCamelCase	)
  							lowerCamelCase_  :    Tuple   	=			output.audios
  
  							lowerCamelCase_  :    Tuple   	=			mel[0, -3:, -3:]
  
  							assert mel_slice.shape == (3, 3)
  							lowerCamelCase_  :    List[str]   	=			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      __a    (		self    :    Any	):
  							return super().test_save_load_local()
  
  
  
  					@skip_mps
  					def      __a    (		self    :    Tuple	):
  							return super().test_dict_tuple_outputs_equivalent()
  
  
  
  					@skip_mps
  					def      __a    (		self    :    Dict	):
  							return super().test_save_load_optional_components()
  
  
  
  					@skip_mps
  					def      __a    (		self    :    str	):
  							return super().test_attention_slicing_forward_pass()
  
  
  
  					def      __a    (		self    :    Any	):
  							pass
  
  
  
  					def      __a    (		self    :    Optional[int]	):
  							pass
  
  
  
  
  
  					@skip_mps
  					def      __a    (		self    :    Optional[Any]	):
  							return super().test_progress_bar()
  
  
  
  
  
  
  @slow
  @require_torch_gpu
  @require_onnxruntime
  @require_note_seq
  class 	UpperCamelCase_     (     unittest.TestCase       ):
  
  
  
  					def      __a    (		self    :    Optional[int]	):
  							# clean up the VRAM after each test
  							super().tearDown()
  							gc.collect()
  							torch.cuda.empty_cache()
  
  
  
  					def      __a    (		self    :    str	):
  							# TODO - test that pipeline can decode tokens in a callback
  							# so that music can be played live
  							lowerCamelCase_  :    Optional[Any]   	=			torch_device
  
  							lowerCamelCase_  :    Union[str, Any]   	=			SpectrogramDiffusionPipeline.from_pretrained('google/music-spectrogram-diffusion'	)
  							lowerCamelCase_  :    str   	=			pipe.melgan
  							lowerCamelCase_  :    Optional[int]   	=			None
  
  							lowerCamelCase_  :    str   	=			pipe.to(lowerCamelCase	)
  							pipe.set_progress_bar_config(disable=lowerCamelCase	)
  
  							def callback(lowerCamelCase    :    int					,			lowerCamelCase    :    int	):
  									# decode mel to audio
  									lowerCamelCase_  :    Union[str, Any]   	=			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
  
  							lowerCamelCase_  :    Any   	=			MidiProcessor()
  							lowerCamelCase_  :    str   	=			processor(lowerCamelCase	)
  
  							lowerCamelCase_  :    str   	=			input_tokens[:3]
  							lowerCamelCase_  :    Dict   	=			torch.manual_seed(0	)
  							pipe(lowerCamelCase					,			num_inference_steps=5					,			generator=lowerCamelCase					,			callback=lowerCamelCase					,			output_type='mel'	)
  
  
  
  					def      __a    (		self    :    str	):
  							lowerCamelCase_  :    List[str]   	=			torch_device
  
  							lowerCamelCase_  :    List[str]   	=			SpectrogramDiffusionPipeline.from_pretrained('google/music-spectrogram-diffusion'	)
  							lowerCamelCase_  :    Union[str, Any]   	=			pipe.to(lowerCamelCase	)
  							pipe.set_progress_bar_config(disable=lowerCamelCase	)
  							lowerCamelCase_  :    int   	=			MidiProcessor()
  
  							lowerCamelCase_  :    Dict   	=			processor(lowerCamelCase	)
  							# just run two denoising loops
  							lowerCamelCase_  :    Optional[Any]   	=			input_tokens[:2]
  
  							lowerCamelCase_  :    List[Any]   	=			torch.manual_seed(0	)
  							lowerCamelCase_  :    str   	=			pipe(lowerCamelCase					,			num_inference_steps=2					,			generator=lowerCamelCase	)
  
  							lowerCamelCase_  :    Dict   	=			output.audios[0]
  
  							assert abs(np.abs(lowerCamelCase	).sum() - 3_612.841	) < 1E-1
  
  
  
  
  
  					def      __a    (		self    :    List[Any]	):
  							lowerCamelCase_  :    int   	=			torch_device
  
  							lowerCamelCase_  :    str   	=			SpectrogramDiffusionPipeline.from_pretrained('google/music-spectrogram-diffusion'	)
  							lowerCamelCase_  :    Optional[int]   	=			pipe.to(lowerCamelCase	)
  							pipe.set_progress_bar_config(disable=lowerCamelCase	)
  
  							lowerCamelCase_  :    Tuple   	=			MidiProcessor()
  
  							lowerCamelCase_  :    int   	=			processor(lowerCamelCase	)
  
  							# just run 4 denoising loops
  							lowerCamelCase_  :    Optional[Any]   	=			input_tokens[:4]
  
  							lowerCamelCase_  :    List[Any]   	=			torch.manual_seed(0	)
  							lowerCamelCase_  :    int   	=			pipe(lowerCamelCase					,			num_inference_steps=1_00					,			generator=lowerCamelCase	)
  
  							lowerCamelCase_  :    List[Any]   	=			output.audios[0]
  							assert abs(np.abs(lowerCamelCase	).sum() - 9_389.1_111	) < 5E-2
  