



# Copyright 2023 The HuggingFace Inc. team. All rights reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
import ast
import difflib
from collections.abc import Mapping
from typing import Any, Callable, Dict







class      __lowerCAmelCase       (     A		):
			pass





def 				A      (	_UpperCAmelCase							:  str				,  _UpperCAmelCase							:  Dict[str, Callable]				,  _UpperCAmelCase							:  Optional[int]=None				,  _UpperCAmelCase							:  Dict=False       )							->					Optional[Any]:







			'''simple docstring'''
			try:
						_UpperCAmelCase									=  ast.parse(_UpperCAmelCase       )
			except SyntaxError as e:
						print('The code generated by the agent is not valid.\n'				,  _UpperCAmelCase       )
						return
			if state is None:
						_UpperCAmelCase									=  {}
			_UpperCAmelCase									=  None
			for idx, node in enumerate(expression.body       ):
						try:
									_UpperCAmelCase									=  evaluate_ast(_UpperCAmelCase				,  _UpperCAmelCase				,  _UpperCAmelCase       )
						except InterpretorError as e:
									_UpperCAmelCase									=  F"Evaluation of the code stopped at line {idx} before the end because of the following error"
									if chat_mode:
												msg += (
												    F". Copy paste the following error message and send it back to the agent:\nI get an error: '{e}'"
												)
									else:
												msg += F":\n{e}"
									print(_UpperCAmelCase       )
									break
						if line_result is not None:
									_UpperCAmelCase									=  line_result

			return result





def 				A      (	_UpperCAmelCase							:  ast.AST				,  _UpperCAmelCase							:  Dict[str, Any]				,  _UpperCAmelCase							:  Dict[str, Callable]       )							->					List[str]:







			'''simple docstring'''
			if isinstance(_UpperCAmelCase				,  ast.Assign       ):
						# Assignement -> we evaluate the assignement which should update the state
						# We return the variable assigned as it may be used to determine the final result.
						return evaluate_assign(_UpperCAmelCase				,  _UpperCAmelCase				,  _UpperCAmelCase       )
			elif isinstance(_UpperCAmelCase				,  ast.Call       ):
						# Function call -> we return the value of the function call
						return evaluate_call(_UpperCAmelCase				,  _UpperCAmelCase				,  _UpperCAmelCase       )
			elif isinstance(_UpperCAmelCase				,  ast.Constant       ):
						# Constant -> just return the value
						return expression.value
			elif isinstance(_UpperCAmelCase				,  ast.Dict       ):
						# Dict -> evaluate all keys and values
						_UpperCAmelCase									=  [evaluate_ast(_UpperCAmelCase				,  _UpperCAmelCase				,  _UpperCAmelCase       ) for k in expression.keys]
						_UpperCAmelCase									=  [evaluate_ast(_UpperCAmelCase				,  _UpperCAmelCase				,  _UpperCAmelCase       ) for v in expression.values]
						return dict(zip(_UpperCAmelCase				,  _UpperCAmelCase       )       )
			elif isinstance(_UpperCAmelCase				,  ast.Expr       ):
						# Expression -> evaluate the content
						return evaluate_ast(expression.value				,  _UpperCAmelCase				,  _UpperCAmelCase       )
			elif isinstance(_UpperCAmelCase				,  ast.For       ):
						# For loop -> execute the loop
						return evaluate_for(_UpperCAmelCase				,  _UpperCAmelCase				,  _UpperCAmelCase       )
			elif isinstance(_UpperCAmelCase				,  ast.FormattedValue       ):
						# Formatted value (part of f-string) -> evaluate the content and return
						return evaluate_ast(expression.value				,  _UpperCAmelCase				,  _UpperCAmelCase       )
			elif isinstance(_UpperCAmelCase				,  ast.If       ):
						# If -> execute the right branch
						return evaluate_if(_UpperCAmelCase				,  _UpperCAmelCase				,  _UpperCAmelCase       )
			elif hasattr(_UpperCAmelCase				,  'Index'       ) and isinstance(_UpperCAmelCase				,  ast.Index       ):
						return evaluate_ast(expression.value				,  _UpperCAmelCase				,  _UpperCAmelCase       )
			elif isinstance(_UpperCAmelCase				,  ast.JoinedStr       ):
						return "".join([str(evaluate_ast(_UpperCAmelCase				,  _UpperCAmelCase				,  _UpperCAmelCase       )       ) for v in expression.values]       )
			elif isinstance(_UpperCAmelCase				,  ast.List       ):
						# List -> evaluate all elements
						return [evaluate_ast(_UpperCAmelCase				,  _UpperCAmelCase				,  _UpperCAmelCase       ) for elt in expression.elts]
			elif isinstance(_UpperCAmelCase				,  ast.Name       ):
						# Name -> pick up the value in the state
						return evaluate_name(_UpperCAmelCase				,  _UpperCAmelCase				,  _UpperCAmelCase       )
			elif isinstance(_UpperCAmelCase				,  ast.Subscript       ):
						# Subscript -> return the value of the indexing
						return evaluate_subscript(_UpperCAmelCase				,  _UpperCAmelCase				,  _UpperCAmelCase       )
			else:
						# For now we refuse anything else. Let's add things as we need them.
						raise InterpretorError(F"{expression.__class__.__name__} is not supported."       )





def 				A      (	_UpperCAmelCase							:  Any				,  _UpperCAmelCase							:  Optional[int]				,  _UpperCAmelCase							:  List[Any]       )							->					Tuple:







			'''simple docstring'''
			_UpperCAmelCase									=  assign.targets
			_UpperCAmelCase									=  evaluate_ast(assign.value				,  _UpperCAmelCase				,  _UpperCAmelCase       )

			if len(_UpperCAmelCase       ) == 1:
						_UpperCAmelCase									=  result
			else:
						if len(_UpperCAmelCase       ) != len(_UpperCAmelCase       ):
									raise InterpretorError(F"Expected {len(_UpperCAmelCase       )} values but got {len(_UpperCAmelCase       )}."       )
						for var_name, r in zip(_UpperCAmelCase				,  _UpperCAmelCase       ):
									_UpperCAmelCase									=  r
			return result





def 				A      (	_UpperCAmelCase							:  Union[str, Any]				,  _UpperCAmelCase							:  Union[str, Any]				,  _UpperCAmelCase							:  int       )							->					int:







			'''simple docstring'''
			if not isinstance(call.func				,  ast.Name       ):
						raise InterpretorError(
						    F"It is not permitted to evaluate other functions than the provided tools (tried to execute {call.func} of "
						    F"type {type(call.func       )}."       )
			_UpperCAmelCase									=  call.func.id
			if func_name not in tools:
						raise InterpretorError(
						    F"It is not permitted to evaluate other functions than the provided tools (tried to execute {call.func.id})."       )

			_UpperCAmelCase									=  tools[func_name]
			# Todo deal with args
			_UpperCAmelCase									=  [evaluate_ast(_UpperCAmelCase				,  _UpperCAmelCase				,  _UpperCAmelCase       ) for arg in call.args]
			_UpperCAmelCase									=  {keyword.arg: evaluate_ast(keyword.value				,  _UpperCAmelCase				,  _UpperCAmelCase       ) for keyword in call.keywords}
			return func(*_UpperCAmelCase				,  **_UpperCAmelCase       )





def 				A      (	_UpperCAmelCase							:  int				,  _UpperCAmelCase							:  Optional[int]				,  _UpperCAmelCase							:  Union[str, Any]       )							->					Union[str, Any]:







			'''simple docstring'''
			_UpperCAmelCase									=  evaluate_ast(subscript.slice				,  _UpperCAmelCase				,  _UpperCAmelCase       )
			_UpperCAmelCase									=  evaluate_ast(subscript.value				,  _UpperCAmelCase				,  _UpperCAmelCase       )
			if isinstance(_UpperCAmelCase				,  (list, tuple)       ):
						return value[int(_UpperCAmelCase       )]
			if index in value:
						return value[index]
			if isinstance(_UpperCAmelCase				,  _UpperCAmelCase       ) and isinstance(_UpperCAmelCase				,  _UpperCAmelCase       ):
						_UpperCAmelCase									=  difflib.get_close_matches(_UpperCAmelCase				,  list(value.keys()       )       )
						if len(_UpperCAmelCase       ) > 0:
									return value[close_matches[0]]

			raise InterpretorError(F"Could not index {value} with '{index}'."       )





def 				A      (	_UpperCAmelCase							:  List[str]				,  _UpperCAmelCase							:  Dict				,  _UpperCAmelCase							:  Optional[int]       )							->					Dict:







			'''simple docstring'''
			if name.id in state:
						return state[name.id]
			_UpperCAmelCase									=  difflib.get_close_matches(name.id				,  list(state.keys()       )       )
			if len(_UpperCAmelCase       ) > 0:
						return state[close_matches[0]]
			raise InterpretorError(F"The variable `{name.id}` is not defined."       )





def 				A      (	_UpperCAmelCase							:  Tuple				,  _UpperCAmelCase							:  Union[str, Any]				,  _UpperCAmelCase							:  Dict       )							->					int:







			'''simple docstring'''
			if len(condition.ops       ) > 1:
						raise InterpretorError('Cannot evaluate conditions with multiple operators'       )

			_UpperCAmelCase									=  evaluate_ast(condition.left				,  _UpperCAmelCase				,  _UpperCAmelCase       )
			_UpperCAmelCase									=  condition.ops[0]
			_UpperCAmelCase									=  evaluate_ast(condition.comparators[0]				,  _UpperCAmelCase				,  _UpperCAmelCase       )

			if isinstance(_UpperCAmelCase				,  ast.Eq       ):
						return left == right
			elif isinstance(_UpperCAmelCase				,  ast.NotEq       ):
						return left != right
			elif isinstance(_UpperCAmelCase				,  ast.Lt       ):
						return left < right
			elif isinstance(_UpperCAmelCase				,  ast.LtE       ):
						return left <= right
			elif isinstance(_UpperCAmelCase				,  ast.Gt       ):
						return left > right
			elif isinstance(_UpperCAmelCase				,  ast.GtE       ):
						return left >= right
			elif isinstance(_UpperCAmelCase				,  ast.Is       ):
						return left is right
			elif isinstance(_UpperCAmelCase				,  ast.IsNot       ):
						return left is not right
			elif isinstance(_UpperCAmelCase				,  ast.In       ):
						return left in right
			elif isinstance(_UpperCAmelCase				,  ast.NotIn       ):
						return left not in right
			else:
						raise InterpretorError(F"Operator not supported: {comparator}"       )





def 				A      (	_UpperCAmelCase							:  Optional[int]				,  _UpperCAmelCase							:  Optional[int]				,  _UpperCAmelCase							:  Union[str, Any]       )							->					Optional[int]:







			'''simple docstring'''
			_UpperCAmelCase									=  None
			if evaluate_condition(if_statement.test				,  _UpperCAmelCase				,  _UpperCAmelCase       ):
						for line in if_statement.body:
									_UpperCAmelCase									=  evaluate_ast(_UpperCAmelCase				,  _UpperCAmelCase				,  _UpperCAmelCase       )
									if line_result is not None:
												_UpperCAmelCase									=  line_result
			else:
						for line in if_statement.orelse:
									_UpperCAmelCase									=  evaluate_ast(_UpperCAmelCase				,  _UpperCAmelCase				,  _UpperCAmelCase       )
									if line_result is not None:
												_UpperCAmelCase									=  line_result
			return result





def 				A      (	_UpperCAmelCase							:  int				,  _UpperCAmelCase							:  Tuple				,  _UpperCAmelCase							:  Optional[Any]       )							->					List[str]:







			'''simple docstring'''
			_UpperCAmelCase									=  None
			_UpperCAmelCase									=  evaluate_ast(for_loop.iter				,  _UpperCAmelCase				,  _UpperCAmelCase       )
			for counter in iterator:
						_UpperCAmelCase									=  counter
						for expression in for_loop.body:
									_UpperCAmelCase									=  evaluate_ast(_UpperCAmelCase				,  _UpperCAmelCase				,  _UpperCAmelCase       )
									if line_result is not None:
												_UpperCAmelCase									=  line_result
			return result





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


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







@unittest.skip('''The note-seq package currently throws an error on import'''		)
class      __lowerCAmelCase       (     A							,		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        _lowerCamelCase			(		self  :    str)  ->       Any:



						"""simple docstring"""






						torch.manual_seed(0)
						_UpperCAmelCase									=  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'					,						)

						_UpperCAmelCase									=  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'					,						)

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

						_UpperCAmelCase									=  DDPMScheduler()

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



			def        _lowerCamelCase			(		self  :    Optional[Any]					,						A  :    str					,						A  :    Union[str, Any]=0)  ->       Optional[Any]:



						"""simple docstring"""






						if str(A).startswith('mps'):
									_UpperCAmelCase									=  torch.manual_seed(A)
						else:
									_UpperCAmelCase									=  torch.Generator(device=A).manual_seed(A)
						_UpperCAmelCase									=  {
						    '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        _lowerCamelCase			(		self  :    str)  ->       Optional[int]:



						"""simple docstring"""






						_UpperCAmelCase									=  'cpu'  # ensure determinism for the device-dependent torch.Generator
						_UpperCAmelCase									=  self.get_dummy_components()
						_UpperCAmelCase									=  SpectrogramDiffusionPipeline(**A)
						_UpperCAmelCase									=  pipe.to(A)
						pipe.set_progress_bar_config(disable=A)

						_UpperCAmelCase									=  self.get_dummy_inputs(A)
						_UpperCAmelCase									=  pipe(**A)
						_UpperCAmelCase									=  output.audios

						_UpperCAmelCase									=  mel[0, -3:, -3:]

						assert mel_slice.shape == (3, 3)
						_UpperCAmelCase									=  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        _lowerCamelCase			(		self  :    Optional[Any])  ->       Optional[Any]:



						"""simple docstring"""






						return super().test_save_load_local()



			@skip_mps
			def        _lowerCamelCase			(		self  :    Optional[Any])  ->       List[str]:



						"""simple docstring"""






						return super().test_dict_tuple_outputs_equivalent()



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



						"""simple docstring"""






						return super().test_save_load_optional_components()



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



						"""simple docstring"""






						return super().test_attention_slicing_forward_pass()



			def        _lowerCamelCase			(		self  :    List[Any])  ->       Dict:



						"""simple docstring"""






						pass



			def        _lowerCamelCase			(		self  :    str)  ->       Any:



						"""simple docstring"""






						pass



			@skip_mps
			def        _lowerCamelCase			(		self  :    Optional[int])  ->       int:



						"""simple docstring"""






						return super().test_progress_bar()







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



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



						"""simple docstring"""






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



			def        _lowerCamelCase			(		self  :    int)  ->       Tuple:



						"""simple docstring"""






						_UpperCAmelCase									=  torch_device

						_UpperCAmelCase									=  SpectrogramDiffusionPipeline.from_pretrained('google/music-spectrogram-diffusion')
						_UpperCAmelCase									=  pipe.melgan
						_UpperCAmelCase									=  None

						_UpperCAmelCase									=  pipe.to(A)
						pipe.set_progress_bar_config(disable=A)

						def callback(A  :    Dict					,						A  :    Optional[int]):
									# decode mel to audio
									_UpperCAmelCase									=  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

						_UpperCAmelCase									=  MidiProcessor()
						_UpperCAmelCase									=  processor(A)

						_UpperCAmelCase									=  input_tokens[:3]
						_UpperCAmelCase									=  torch.manual_seed(0)
						pipe(A					,						num_inference_steps=5					,						generator=A					,						callback=A					,						output_type='mel')



			def        _lowerCamelCase			(		self  :    str)  ->       Optional[int]:



						"""simple docstring"""






						_UpperCAmelCase									=  torch_device

						_UpperCAmelCase									=  SpectrogramDiffusionPipeline.from_pretrained('google/music-spectrogram-diffusion')
						_UpperCAmelCase									=  pipe.to(A)
						pipe.set_progress_bar_config(disable=A)
						_UpperCAmelCase									=  MidiProcessor()

						_UpperCAmelCase									=  processor(A)
						# just run two denoising loops
						_UpperCAmelCase									=  input_tokens[:2]

						_UpperCAmelCase									=  torch.manual_seed(0)
						_UpperCAmelCase									=  pipe(A					,						num_inference_steps=2					,						generator=A)

						_UpperCAmelCase									=  output.audios[0]

						assert abs(np.abs(A).sum() - 3_6_1_2.8_4_1) < 1E-1



			def        _lowerCamelCase			(		self  :    List[str])  ->       Tuple:



						"""simple docstring"""






						_UpperCAmelCase									=  torch_device

						_UpperCAmelCase									=  SpectrogramDiffusionPipeline.from_pretrained('google/music-spectrogram-diffusion')
						_UpperCAmelCase									=  pipe.to(A)
						pipe.set_progress_bar_config(disable=A)

						_UpperCAmelCase									=  MidiProcessor()

						_UpperCAmelCase									=  processor(A)

						# just run 4 denoising loops
						_UpperCAmelCase									=  input_tokens[:4]

						_UpperCAmelCase									=  torch.manual_seed(0)
						_UpperCAmelCase									=  pipe(A					,						num_inference_steps=1_00					,						generator=A)

						_UpperCAmelCase									=  output.audios[0]
						assert abs(np.abs(A).sum() - 9_3_8_9.1_1_1_1) < 5E-2
