УПРОСТИТЕ ЭТОТ КОД, ЧТОБЫ СТАЛ ПОЛНОСТЬЮ РАБОЧИМ: import os
import logging
import openai
import time
import sys
import lxml.etree as ET
import librosa
import noisereduce as nr
import numpy as np
import torch
import asyncio
import soundfile as sf
from torchaudio.transforms import MelSpectrogram
from transformers import Wav2Vec2Tokenizer, Wav2Vec2ForCTC, HubertForCTC
from music21 import stream, note, converter, instrument

# Настройка логирования
logging.basicConfig(level=logging.INFO)

# Загрузка ключа API из переменных среды
openai.api_key = os.environ.get(‘OPENAI_API_KEY’)

# Проверка, что ключ API загружен
if openai.api_key:
try:
# Пробный запрос к OpenAI API для проверки подключения
response = openai.Completion.create(
model=“gpt-4”,
prompt=“Hello, world!”,
max_tokens=5
)
logging.info(“Подключение к OpenAI API успешно”)
except openai.error.AuthenticationError:
logging.error(“Ошибка аутентификации: Неверный ключ OpenAI API”)
except Exception as e:
logging.error(f"Ошибка при подключении к OpenAI API: {e}“)
else:
logging.warning(“Ключ OpenAI API не найден. API будет проигнорирован”)

# Определение функции для предварительной обработки аудио
async def preprocess_audio(file_path):
try:
logging.info(f"Loading and preprocessing audio from {file_path}”)
y, sr = librosa.load(file_path, sr=None)
y = librosa.util.normalize(y)
y = nr.reduce_noise(y=y, sr=sr)
y = np.asfortranarray(y)
logging.info(“Audio preprocessing completed”)
return y, sr
except Exception as e:
logging.error(f"Error preprocessing audio: {e}“)
return None, None

# Определение функции для извлечения признаков
async def extract_features(y, sr):
try:
logging.info(“Extracting features”)
mel_spec = MelSpectrogram()(torch.tensor(y).unsqueeze(0))
logging.info(“Feature extraction completed”)
return mel_spec
except Exception as e:
logging.error(f"Error extracting features: {e}”)
return None

# Определение функции для создания и сохранения MusicXML файла
def save_musicxml(score, output_path):
try:
logging.info(f"Saving MusicXML file to {output_path}“)
tree = ET.ElementTree(score)
tree.write(output_path, encoding=‘UTF-8’, xml_declaration=True, pretty_print=True)
logging.info(f"MusicXML file saved successfully: {output_path}”)
except Exception as e:
logging.error(f"Error saving MusicXML file: {e}“)

# Пример функции для анализа аудио
def analyze_audio(y, sr):
try:
logging.info(“Analyzing audio”)
tempo, beat_frames = librosa.beat.beat_track(y=y, sr=sr)
logging.info(f"Audio analysis completed: tempo={tempo}, beat_frames={beat_frames}”)
except Exception as e:
logging.error(f"Error analyzing audio: {e}“)

# Пример функции для генерации текстовых описаний
def generate_text_descriptions(y, sr):
try:
logging.info(“Generating text descriptions”)
description = f"Audio has a sample rate of {sr} and length of {len(y)} samples.”
logging.info(f"Text descriptions generated: {description}“)
except Exception as e:
logging.error(f"Error generating text descriptions: {e}”)

# Пример функции для визуализации партитуры
def visualize_score(score, output_image_path):
try:
logging.info(“Visualizing score”)
tree = ET.ElementTree(score)
tree.write(output_image_path, encoding=‘UTF-8’, xml_declaration=True, pretty_print=True)
logging.info(f"Score visualization completed: {output_image_path}“)
except Exception as e:
logging.error(f"Error visualizing score: {e}”)

# Пример функции для сепарации источников
def separate_sources(y, sr):
try:
logging.info(“Separating sources”)
# Пример кода для сепарации источников
vocals = y[:len(y)//2]
accompaniment = y[len(y)//2:]
logging.info(“Source separation completed”)
except Exception as e:
logging.error(f"Error separating sources: {e}“)

# Пример функции для оптимизации аудио
def optimize_audio(y, sr):
try:
logging.info(“Optimizing audio”)
optimized_audio = nr.reduce_noise(y=y, sr=sr)
logging.info(“Audio optimization completed”)
except Exception as e:
logging.error(f"Error optimizing audio: {e}”)

# Определение функции main
async def main(file_path, output_path, output_image_path):
try:
# Проверка существования директории для сохранения файла
output_dir = os.path.dirname(output_path)
if not os.path.exists(output_dir):
logging.error(f"Output directory ‘{output_dir}’ does not exist. MusicXML file will not be saved.“)
sys.exit(1)

# Проверка существования файла
if os.path.exists(output_path):
logging.info(f"MusicXML file ‘{output_path}’ will be overwritten.”)
else:
logging.info(f"MusicXML file will be saved to ‘{output_path}’.“)

# Предварительная обработка аудио
y, sr = await preprocess_audio(file_path)
if y is None or sr is None:
logging.error(“Error in audio preprocessing step”)
return

# Извлечение признаков
features = await extract_features(y, sr)
if features is None:
logging.error(“Error in feature extraction step”)
return

# Создание простого XML-документа
score = ET.Element(“score-partwise”, version=“3.1”)
part = ET.SubElement(score, “part”, id=“P1”)
measure = ET.SubElement(part, “measure”, number=“1”)
note_element = ET.SubElement(measure, “note”)
pitch_element = ET.SubElement(note_element, “pitch”)
ET.SubElement(pitch_element, “step”).text = “C”
ET.SubElement(pitch_element, “octave”).text = “4”
ET.SubElement(note_element, “duration”).text = “1”
ET.SubElement(note_element, “type”).text = “quarter”

# Сохранение файла
save_musicxml(score, output_path)

# Дополнительные шаги
logging.info(“Дополнительные шаги выполнены”)

# 1. Анализ аудио
logging.info(“Starting audio analysis…”)
analyze_audio(y, sr)
logging.info(“Audio analysis completed”)

# 2. Генерация текстовых описаний
logging.info(“Generating text descriptions…”)
generate_text_descriptions(y, sr)
logging.info(“Text descriptions generated”)

# 3. Визуализация партитуры
logging.info(“Visualizing score…”)
visualize_score(score, output_image_path)
logging.info(“Score visualization completed”)

# 4. Сепарация источников
logging.info(“Starting source separation…”)
separate_sources(y, sr)
logging.info(“Source separation completed”)

# 5. Оптимизация аудио
logging.info(“Optimizing audio…”)
optimize_audio(y, sr)
logging.info(“Audio optimization completed”)

except Exception as e:
logging.error(f"Error in main function: {e}”)
sys.exit(1)

from torch import nn, optim
from torch.utils.data import DataLoader, Dataset
import torchaudio
import crepe
import madmom
from music21 import stream, note, converter, instrument
from scipy import signal
import openunmix
from demucs import pretrained
from demucs.apply import apply_model
import openl3
import pydub
import parselmouth as pm
from spleeter.separator import Separator
import tonic
from diskcache import Cache
import tensorflow as tf
import importlib.util
import threading
import sounddevice as sd
from pydub import AudioSegment
from pydub.playback import play

# Проверка наличия GPU с помощью TensorFlow
if tf.test.gpu_device_name():
print(‘GPU found’)
else:
print(‘No GPU found, using CPU instead’)

# Добавление пути для импорта onsets_and_frames
sys.path.append(r’C:\Users\ARTYOM\onsets-and-frames\onsets_and_frames’)

# Проверка наличия файла example.py
example_path = r’C:\Users\ARTYOM\onsets-and-frames\onsets_and_frames\example.py’
if not os.path.exists(example_path):
logging.error(“Файл example.py не найден в директории onsets_and_frames”)
sys.exit(1)

# Импорт необходимых модулей из onsets_and_frames с использованием importlib
try:
spec = importlib.util.spec_from_file_location(“example”, example_path)
example = importlib.util.module_from_spec(spec)
spec.loader.exec_module(example)
NonBinaryClassification = example.NonBinaryClassification
logging.info(“Successfully imported NonBinaryClassification from onsets_and_frames.example”)
except Exception as e:
logging.error(f"Error importing NonBinaryClassification from onsets_and_frames.example: {e}“)
sys.exit(1)

# Загрузка ключа API из переменных среды
openai.api_key = os.environ.get(‘OPENAI_API_KEY’)

# Настройка логирования
logging.basicConfig(level=logging.INFO)

# Настройка кэширования
cache = Cache(‘audio_cache’)

# Конфигурационные параметры
config = {
“output_dir”: “output”,
“noise_reduction_level”: 5,
“pitch_shift”: 2,
“speed_change”: 1.5,
“batch_size”: 16,
“epochs”: 10,
“target_format”: “wav”
}

# Определение функции custom_pad_center
def custom_pad_center(data, size):
“””
Центрирует и заполняет массив до указанного размера.
“”“
if len(data) >= size:
return data
pad_size = (size - len(data)) // 2
return np.pad(data, (pad_size, size - len(data) - pad_size), mode=‘constant’)

# Определение функции melspectrogram
def melspectrogram(y, sr, n_mels=128, fmax=8000):
S = librosa.feature.melspectrogram(y=y, sr=sr, n_mels=n_mels, fmax=fmax)
return librosa.power_to_db(S, ref=np.max)

# Асинхронная функция для конвейера обработки
async def preprocess_audio(file_path):
“””
Конвейер обработки аудио
“”“
try:
logging.info(f"Loading and preprocessing audio from {file_path}”)
y, sr = librosa.load(file_path, sr=None)

if (y, sr) in cache:
logging.info(“Using cached preprocessed audio”)
return cache[(y, sr)]

y = librosa.util.normalize(y)
y = nr.reduce_noise(y=y, sr=sr)
y = signal.wiener(y)

cache[(y, sr)] = (y, sr)
logging.info(“Audio preprocessing completed”)

return y, sr
except Exception as e:
logging.error(f"Error preprocessing audio: {e}“)
return None, None

async def extract_features_task(y, sr):
“””
Асинхронная задача для извлечения признаков
“”“
try:
cqt = librosa.cqt(y, sr=sr)
cqt_db = librosa.amplitude_to_db(np.abs(cqt), ref=np.max)

# Pitch detection using CREPE
_, frequency, confidence, _ = crepe.predict(y, sr)

return cqt_db, frequency, confidence
except Exception as e:
logging.error(f"Error extracting features: {e}”)
return None, None, None

def transformer_features_wav2vec2(y, sr):
“”“
Асинхронная задача для извлечения признаков с использованием Wav2Vec2
“””
try:
tokenizer = Wav2Vec2Tokenizer.from_pretrained(“facebook/wav2vec2-base-960h”)
model = Wav2Vec2ForCTC.from_pretrained(“facebook/wav2vec2-base-960h”)

input_values = tokenizer(y.tolist(), return_tensors=“pt”, padding=True).input_values
logits = model(input_values).logits
predicted_ids = torch.argmax(logits, dim=-1)

transcription = tokenizer.batch_decode(predicted_ids)[0]
return transcription
except Exception as e:
logging.error(f"Error extracting transformer features with Wav2Vec2: {e}“)
return None

def transformer_features_hubert(y, sr):
“””
Асинхронная задача для извлечения признаков с использованием HuBERT
“”“
try:
tokenizer = Wav2Vec2Tokenizer.from_pretrained(“facebook/wav2vec2-base-960h”)
model = HubertForCTC.from_pretrained(“facebook/hubert-large-ls960-ft”)

input_values = tokenizer(y.tolist(), return_tensors=“pt”, padding=True).input_values
logits = model(input_values).logits
predicted_ids = torch.argmax(logits, dim=-1)

transcription = tokenizer.batch_decode(predicted_ids)[0]
return transcription
except Exception as e:
logging.error(f"Error extracting transformer features with Hubert: {e}”)
return None

def generate_text_descriptions_with_chatgpt(prompt):
“”“
Генерация текста с использованием GPT-4
“””
try:
response = openai.Completion.create(
model=“gpt-4”,
prompt=prompt,
max_tokens=150 # Увеличиваем количество токенов для получения более детального ответа
)
return response.choices[0].text.strip()
except Exception as e:
logging.error(f"Error generating text descriptions with GPT-4: {e}“)
return None

class AudioPipeline:
def init(self, config):
self.config = config

def play_audio(self, audio):
try:
data = np.array(audio.get_array_of_samples())
sr = audio.frame_rate
sd.play(data, sr)
sd.wait()
except Exception as e:
logging.error(f"Error playing audio: {e}”)

async def run_pipeline(self, file_path):
“”“
Запуск конвейера обработки
“””
# Предварительная обработка
y, sr = await preprocess_audio(file_path)
if y is None or sr is None:
logging.error(“Error in audio preprocessing step”)
return

# Запуск задач извлечения признаков в фоновом режиме
features_task = extract_features_task(y, sr)

# Запуск Преобразующих Моделей
transcription_wav2vec2_task = asyncio.to_thread(transformer_features_wav2vec2, y, sr)
transcription_hubert_task = asyncio.to_thread(transformer_features_hubert, y, sr)

# Ожидание завершения задач
features_result = await features_task
transcription_wav2vec2 = await transcription_wav2vec2_task
transcription_hubert = await transcription_hubert_task

cqt_db, frequency, confidence = features_result

logging.info(f"Feature extraction complete: {cqt_db.shape}, {len(frequency)}, {transcription_wav2vec2}, {transcription_hubert}“)

# Запуск дополнительных задач
additional_features = await asyncio.to_thread(self.additional_features_essentia, y, sr)

# Сепарация источников
sources_openunmix = await asyncio.to_thread(self.source_separation_openunmix, y, sr)
sources_demucs_v3 = await asyncio.to_thread(self.source_separation_demucs_v3, y, sr)
vocals_path, accompaniment_path = await asyncio.to_thread(self.source_separation_spleeter, file_path)

logging.info(f"Source separation complete: OpenUnmix: {sources_openunmix}, Demucs V3: {sources_demucs_v3}, Spleeter: {vocals_path}, {accompaniment_path}”)

# Анализ высоты тона с использованием Parselmouth
pitch_analysis = await asyncio.to_thread(self.analyze_pitch_with_parselmouth, file_path)
logging.info(f"Pitch analysis (Parselmouth): {pitch_analysis}“)

# Оптимизация
audio = AudioSegment.from_file(file_path)
reduced_noise_audio = self.reduce_noise(audio, self.config[“noise_reduction_level”])
modified_audio = self.change_timbre_and_speed(reduced_noise_audio, self.config[“pitch_shift”], self.config[“speed_change”])
normalized_audio = self.normalize_volume_level(modified_audio)

# Использование OpenAI для предварительной обработки аудио
optimized_audio = self.optimize_audio_with_openai(y, sr)

return (cqt_db, frequency, confidence, transcription_wav2vec2, transcription_hubert, additional_features,
sources_openunmix, sources_demucs_v3, vocals_path, accompaniment_path, pitch_analysis, normalized_audio, optimized_audio)

def reduce_noise(self, audio, noise_reduction_level=5):
try:
reduced_noise_audio = audio - noise_reduction_level
logging.info(“Noise reduced”)
return reduced_noise_audio
except Exception as e:
logging.error(f"Error reducing noise: {e}”)
return None

def change_timbre_and_speed(self, audio, pitch_shift=2, speed_change=1.5):
try:
new_timbre_audio = audio._spawn(audio.raw_data, overrides={‘frame_rate’: int(audio.frame_rate * pitch_shift)})
new_timbre_speed_audio = new_timbre_audio.speedup(playback_speed=speed_change)
logging.info(“Timbre and speed changed”)
return new_timbre_speed_audio
except Exception as e:
logging.error(f"Error changing timbre and speed: {e}“)
return None

def normalize_volume_level(self, audio):
try:
normalized_audio = audio.apply_gain(-audio.dBFS)
logging.info(“Volume level normalized”)
return normalized_audio
except Exception as e:
logging.error(f"Error normalizing volume level: {e}”)
return None

def analyze_pitch_with_parselmouth(self, file_path):
try:
snd = parselmouth.Sound(file_path)
pitch = snd.to_pitch()
pitch_values = pitch.selected_array[‘frequency’]
return pitch_values
except Exception as e:
logging.error(f"Error analyzing pitch with Parselmouth: {e}“)
return None

def additional_features_essentia(self, y, sr):
try:
from essentia.standard import MusicExtractor, MonoLoader
loader = MonoLoader()
y = loader(y)
extractor = MusicExtractor(lowlevelStats=[‘mean’, ‘stdev’])
features, _ = extractor(y)
return features
except Exception as e:
logging.error(f"Error extracting additional features with Essentia: {e}”)
return None

def source_separation_openunmix(self, y, sr):
try:
separator = openunmix.separate(
audio_signal=y.tolist(),
sample_rate=sr,
targets=[“vocals”, “drums”, “bass”, “other”]
)
return {
“vocals”: separator[“vocals”],
“drums”: separator[“drums”],
“bass”: separator[“bass”],
“other”: separator[“other”],
}
except Exception as e:
logging.error(f"Error in source separation with OpenUnmix: {e}“)
return None

def source_separation_demucs_v3(self, y, sr):
try:
model = pretrained.get_model(‘htdemucs’)
model.cpu()
wav = torch.tensor(y).unsqueeze(0)
sources = apply_model(model, wav, device=‘cpu’, split=True)
return sources
except Exception as e:
logging.error(f"Error in source separation with Demucs v3: {e}”)
return None

def source_separation_spleeter(self, file_path):
try:
separator = Separator(‘spleeter:2stems’)
separator.separate_to_file(file_path, output_path=‘output’)
return ‘output/vocals.wav’, ‘output/accompaniment.wav’
except Exception as e:
logging.error(f"Error in source separation with Spleeter: {e}“)
return None, None

def recognize_instruments(self, y, sr):
try:
import yamnet as yamnet_model
params = yamnet_model.Params()
yamnet = yamnet_model.yamnet(params)
yamnet.load_weights(‘yamnet.h5’)

waveform = np.reshape(y, [len(y), 1])

class_scores, embeddings, spectrogram = yamnet(waveform)

classes = yamnet_model.class_names(params)
top_classes = np.argsort(class_scores, axis=-1)[0, -5:][::-1]
recognized_classes = [classes[i] for i in top_classes]

return recognized_classes
except Exception as e:
logging.error(f"Error recognizing instruments: {e}”)
return None

def recognize_genre(self, y, sr):
try:
features = self.additional_features_essentia(y, sr)
genre_description = generate_text_descriptions_with_chatgpt(f"Recognize the genre based on these features: {features}“)
return genre_description
except Exception as e:
logging.error(f"Error recognizing genre: {e}”)
return None

# Функция для создания MusicXML
import lxml.etree as ET # Для ручной обработки XML

def create_musicxml(frequencies, beats, frames, onsets, offsets, text_descriptions, instruments, use_music21=True, use_lxml=True):
score = None

if use_music21:
try:
score = stream.Score()
part = stream.Part()

for freq, beat, frame, onset, offset, description, inst in zip(frequencies, beats, frames, onsets, offsets, text_descriptions, instruments):
if onset >= 0:
n = note.Note()
n.pitch.frequency = freq
n.quarterLength = beat
n.offset = offset
n.addLyric(description)

# Добавление инструмента к ноте
if hasattr(instrument, inst):
instrument_class = getattr(instrument, inst)
part.append(instrument_class())
else:
n.addLyric(“Unknown Instrument”)

part.append(n)

score.append(part)
except Exception as e:
logging.error(f"Error creating MusicXML with music21: {e}“)

if use_lxml:
try:
# Создание MusicXML с использованием lxml. Это упрощенная версия и требует вашего расширения.
score = ET.Element(“score-partwise”, version=“3.1”)
part = ET.SubElement(score, “part”, id=“P1”)
measure = ET.SubElement(part, “measure”, number=“1”)

for freq, beat, frame, onset, offset, description, inst in zip(frequencies, beats, frames, onsets, offsets, text_descriptions, instruments):
if onset >= 0:
note_element = ET.SubElement(measure, “note”)
pitch_element = ET.SubElement(note_element, “pitch”)
ET.SubElement(pitch_element, “step”).text = “C” # Пример, вам нужно будет преобразовать частоту в ноту
ET.SubElement(pitch_element, “octave”).text = “4”
ET.SubElement(note_element, “duration”).text = str(int(beat * 4)) # Просто пример
ET.SubElement(note_element, “type”).text = “quarter” # Вам нужно правильно определить тип нот
lyric_element = ET.SubElement(note_element, “lyric”)
ET.SubElement(lyric_element, “text”).text = description
except Exception as e:
logging.error(f"Error creating MusicXML with lxml: {e}”)

return score

# Функция для сохранения MusicXML
def save_musicxml(score, output_path, use_music21=True):
try:
output_dir = os.path.dirname(output_path)
if not os.path.exists(output_dir):
os.makedirs(output_dir)

if use_music21:
score.write(‘musicxml’, fp=output_path)
else:
tree = ET.ElementTree(score)
tree.write(output_path, encoding=‘UTF-8’, xml_declaration=True, pretty_print=True)

logging.info(f’MusicXML saved to {output_path}‘)

# Проверка, что файл был успешно сохранен
if os.path.exists(output_path):
logging.info(f"MusicXML file saved successfully: {output_path}“)
else:
logging.error(f"Failed to save MusicXML file: {output_path}”)
except Exception as e:
logging.error(f"Error writing MusicXML: {e}“)

def optimize_audio_with_openai(self, y, sr):
“””
Использование OpenAI для оптимизации аудио
“”“
try:
# Анализ аудио с помощью OpenAI
audio_analysis = generate_text_descriptions_with_chatgpt(f"Analyze the audio and provide recommendations for optimal preprocessing, such as noise reduction level, pitch shift, and speed change.”)

# Извлечение рекомендаций из анализа
noise_reduction_level = int(audio_analysis.split(“,”)[0].split(“:”)[1].strip())
pitch_shift = float(audio_analysis.split(“,”)[1].split(“:”)[1].strip())
speed_change = float(audio_analysis.split(“,”)[2].split(“:”)[1].strip())

# Применение рекомендаций
reduced_noise_audio = self.reduce_noise(y, noise_reduction_level)
modified_audio = self.change_timbre_and_speed(reduced_noise_audio, pitch_shift, speed_change)
normalized_audio = self.normalize_volume_level(modified_audio)

return normalized_audio
except Exception as e:
logging.error(f"Error optimizing audio with OpenAI: {e}“)
return None

def visualize_score(score, output_image_path):
“””
Визуализация партитуры и сохранение в виде изображения
“”"
try:
s = converter.parse(score)
s.show(‘text’) # Показать партитуру в текстовом виде (можно удалить, если не нужно)

fp = s.write(‘musicxml.png’, fp=output_image_path)
logging.info(f’Партитура сохранена в {fp}’)
except Exception as e:
logging.error(f"Error visualizing score: {e}“)

class MusicDataset(Dataset):
def init(self, features, labels):
self.features = features
self.labels = labels

def len(self):
return len(self.features)

def getitem(self, idx):
return self.features[idx], self.labels[idx]

def train_cnn_model(train_loader, in_channels, out_channels, epochs=10):
model = AudioCNN(in_channels, out_channels)
criterion = nn.MSELoss()
optimizer = optim.Adam(model.parameters(), lr=0.001)

for epoch in range(epochs):
for inputs, labels in train_loader:
# Возможно, добавить использование GPU
if torch.cuda.is_available():
inputs, labels = inputs.cuda(), labels.cuda()
optimizer.zero_grad()
outputs = model(inputs)
loss = criterion(outputs, labels)
loss.backward()
optimizer.step()

logging.info(f"Epoch {epoch+1}/{epochs}, Loss: {loss.item()}”)

return model

def generate_text_descriptions(features):
try:
response = openai.Completion.create(
model=“gpt-4”,
prompt=f"Generate a coherent description based on the following features: {features}“,
max_tokens=100
)
return response.choices[0].text.strip()
except Exception as e:
logging.error(f"Error generating text descriptions: {e}”)
return None

def modify_notes_with_gpt4(score, action=“add”, instrument=“piano”):
notes_data = [(n.pitch.frequency, n.quarterLength, n.offset) for n in score.flat.notes]
prompt = f"Партитура содержит следующие данные нот:\n{notes_data}\n Пожалуйста, предоставьте необходимые изменения, чтобы {(‘добавить’ if action == ‘add’ else ‘удалить’)} ноты для инструмента ‘{instrument}’, и обеспечьте соблюдение правильных музыкальных правил. Также проанализируйте сложность композиции.“

try:
response = openai.Completion.create(
model=“gpt-4”,
prompt=prompt,
max_tokens=150 # Увеличиваем количество токенов для получения более детального ответа
)

modified_notes = response.choices[0].text.strip().split(‘\n’)

for note_data in modified_notes:
try:
note_info = note_data.split(‘,’)
if len(note_info) == 3:
freq, length, offset = map(float, note_info)
if action == “add”:
new_note = note.Note()
new_note.pitch.frequency = freq
new_note.quarterLength = length
new_note.offset = offset
score.append(new_note)
elif action == “remove”:
for n in score.flat.notes:
if (n.pitch.frequency == freq and
n.quarterLength == length and
n.offset == offset):
score.remove(n)
break
except ValueError:
logging.warning(f"Skipping invalid note data: {note_data}”)
except Exception as e:
logging.error(f"Error generating modifications with GPT-4: {e}“)

return score

def plot_spectrogram(y, sr, output_image_path):
try:
import matplotlib.pyplot as plt
import librosa.display

D = librosa.amplitude_to_db(np.abs(librosa.stft(y)), ref=np.max)
plt.figure(figsize=(12, 8))
librosa.display.specshow(D, sr=sr, x_axis=‘time’, y_axis=‘log’)
plt.colorbar(format=‘%+2.0f dB’)
plt.title(‘Spectrogram’)
plt.savefig(output_image_path)
logging.info(f’Spectrogram saved to {output_image_path}')
plt.close()
except Exception as e:
logging.error(f"Error plotting spectrogram: {e}”)

def plot_chromagram(y, sr, output_image_path):
try:
import matplotlib.pyplot as plt
import librosa.display

chromagram = librosa.feature.chroma_stft(y=y, sr=sr)
plt.figure(figsize=(12, 8))
librosa.display.specshow(chromagram, sr=sr, x_axis=‘time’, y_axis=‘chroma’)
plt.colorbar(format=‘%+2.0’)
plt.title(‘Chromagram’)
plt.savefig(output_image_path)
logging.info(f’Chromagram saved to {output_image_path}‘)
plt.close()
except Exception as e:
logging.error(f"Error plotting chromagram: {e}“)

def compute_spectral_centroid(y, sr):
try:
spectral_centroids = librosa.feature.spectral_centroid(y=y, sr=sr)[0]
return spectral_centroids
except Exception as e:
logging.error(f"Error computing spectral centroid: {e}”)
return None

def compute_tonality_and_stability(y, sr):
try:
chroma_stft = librosa.feature.chroma_stft(y=y, sr=sr)
key = librosa.estimate_tuning(y=y, sr=sr)
return key, chroma_stft
except Exception as e:
logging.error(f"Error computing tonality and stability: {e}“)
return None, None

def harmonic_rhythm_analysis(y, sr):
try:
tempo, beat_frames = librosa.beat.beat_track(y=y, sr=sr)
chords = librosa.effects.harmonic(y)

# Анализ ритма с использованием madmom
proc = madmom.features.beats.RNNBeatProcessor()(y)
beat_times = madmom.features.beats.DBNBeatTrackingProcessor(beats_per_bar=[3, 4])(proc)

return tempo, beat_frames, chords, beat_times
except Exception as e:
logging.error(f"Error in harmonic rhythm analysis: {e}”)
return None, None, None, None

# Реализация потоковой обработки и воспроизведения аудио в реальном времени
async def process_audio_chunk(chunk, sr):
try:
y = np.array(chunk.get_array_of_samples())
y = librosa.util.normalize(y)
y = nr.reduce_noise(y=y, sr=sr)
y = signal.wiener(y)
return y
except Exception as e:
logging.error(f"Error processing audio chunk: {e}“)
return None

async def play_and_process_audio(file_path):
try:
audio = AudioSegment.from_file(file_path)
sr = audio.frame_rate
chunk_size = 1000 # Размер буфера в миллисекундах

for i in range(0, len(audio), chunk_size):
chunk = audio[i:i + chunk_size]
y = await process_audio_chunk(chunk, sr)
if y is not None:
sd.play(y, sr)
sd.wait()
else:
logging.error(“Error processing audio chunk, skipping playback”)
except Exception as e:
logging.error(f"Error in play_and_process_audio: {e}”)

def main(file_path, output_path, output_image_path):
# Конфигурирование кэширования данных
dataset_config = {
“features”: “path_to_features_dataset”,
“labels”: “path_to_labels_dataset”
}

# Инициализация конвейера и запуск обработки
audio_pipeline = AudioPipeline(config)
loop = asyncio.get_event_loop()
results = loop.run_until_complete(audio_pipeline.run_pipeline(file_path))

if results:
# Обработка результатов
(cqt_db, frequency, confidence, transcription_wav2vec2, transcription_hubert, additional_features,
sources_openunmix, sources_demucs_v3, vocals_path, accompaniment_path, pitch_analysis, normalized_audio, optimized_audio) = results

logging.info(f"Processing results: {cqt_db.shape}, {len(frequency)}, {transcription_wav2vec2}, {transcription_hubert}")

# Гармонический анализ и ритм
y, sr = librosa.load(file_path, sr=None)
tempo, beat_frames, chords, beat_times = harmonic_rhythm_analysis(y, sr)
if tempo is None or beat_frames is None or chords is None or beat_times is None:
logging.error(“Error in harmonic rhythm analysis”)
return

# Генерация текстовых описаний музыкальных фрагментов с использованием ChatGPT (GPT-4)
text_descriptions = generate_text_descriptions(additional_features)
if text_descriptions is None:
logging.error(“Error generating text descriptions”)
return

# Распознавание музыкальных инструментов
instruments = audio_pipeline.recognize_instruments(y, sr)
if instruments is None:
logging.error(“Error recognizing instruments”)
return

# Распознавание жанра музыки
genre_description = audio_pipeline.recognize_genre(y, sr)
if genre_description is None:
logging.error(“Error recognizing genre”)
return

# Спектральный центроид
spectral_centroid = compute_spectral_centroid(y, sr)
if spectral_centroid is None:
logging.error(“Error computing spectral centroid”)
return

# Тональность и тональная стабильность
key, chroma_stft = compute_tonality_and_stability(y, sr)
if key is None or chroma_stft is None:
logging.error(“Error computing tonality and stability”)
return

# Создание MusicXML из результатов
# Используем обе библиотеки для преобразования
score_music21 = create_musicxml(frequency, beat_times, range(len(beat_times)), range(len(beat_times)),
range(len(beat_times)), text_descriptions, instruments, use_music21=True, use_lxml=False)
score_lxml = create_musicxml(frequency, beat_times, range(len(beat_times)), range(len(beat_times)),
range(len(beat_times)), text_descriptions, instruments, use_music21=False, use_lxml=True)

if score_music21 is None and score_lxml is None:
logging.error(“Error creating MusicXML”)
return

# Логика добавления или удаления нот с использованием GPT-4 (ChatGPT)
score_music21 = modify_notes_with_gpt4(score_music21, action=“add”, instrument=“piano”)

# Сохранение и визуализация партитуры в формате MusicXML и изображения
output_path_music21 = output_path.replace(’.xml’, ‘_music21.xml’)
output_path_lxml = output_path.replace(‘.xml’, ‘_lxml.xml’)

save_musicxml(score_music21, output_path_music21, use_music21=True)
save_musicxml(score_lxml, output_path_lxml, use_music21=False)

visualize_score(score_music21, output_image_path)

# Визуализация спектрограммы аудиофайла
spectrogram_image_path = output_image_path.replace(“.png”, “_spectrogram.png”)
y, sr = await preprocess_audio(file_path)
plot_spectrogram(y, sr, spectrogram_image_path)

# Визуализация хромограммы аудиофайла
chromagram_image_path = output_image_path.replace(“.png”, “_chromagram.png”)
plot_chromagram(y, sr, chromagram_image_path)

# Демонстрация воспроизведения аудио
audio_pipeline.play_audio(normalized_audio)
audio_pipeline.play_audio(optimized_audio)

# Окончательное описание всех источников
full_description = {
“Transcription (Wav2Vec2)”: transcription_wav2vec2,
“Transcription (HuBERT)”: transcription_hubert,
“Genre Description (GPT-4)”: genre_description,
“Instrument Recognition (YAMNet)”: instruments,
“Additional Features (Essentia)”: additional_features,
“Pitch Analysis (Parselmouth)”: pitch_analysis,
“Spectral Centroid”: spectral_centroid,
“Key (Tuning)”: key,
“Chroma (Tonality Stability)”: chroma_stft
}

logging.info(“Full Description:”)
for key, value in full_description.items():
logging.info(f"{key}: {value}“)

# Сохранение оптимизированного аудио
optimized_audio_path = os.path.join(config[“output_dir”], “optimized_audio.wav”)
sf.write(optimized_audio_path, optimized_audio, sr)
logging.info(f"Optimized audio saved to: {optimized_audio_path}”)

# Сохранение разделенных источников
if vocals_path and accompaniment_path:
logging.info(f"Vocals saved to: {vocals_path}“)
logging.info(f"Accompaniment saved to: {accompaniment_path}”)

# Сохранение дополнительных визуализаций
plot_spectral_features(y, sr, output_image_path.replace(“.png”, “_spectral_features.png”))
plot_harmonic_rhythm(tempo, beat_times, chords, output_image_path.replace(“.png”, “_harmonic_rhythm.png”))

# Сохранение дополнительных данных
save_additional_data(
file_path,
cqt_db, frequency, confidence,
transcription_wav2vec2, transcription_hubert,
additional_features, instruments, genre_description,
pitch_analysis, spectral_centroid, key, chroma_stft,
output_dir=config[“output_dir”]
)

def plot_spectral_features(y, sr, output_image_path):
try:
import matplotlib.pyplot as plt
import librosa.display

# Спектрограмма
D = librosa.amplitude_to_db(np.abs(librosa.stft(y)), ref=np.max)
plt.figure(figsize=(12, 8))
librosa.display.specshow(D, sr=sr, x_axis=‘time’, y_axis=‘log’)
plt.colorbar(format=‘%+2.0f dB’)
plt.title(‘Spectrogram’)
plt.savefig(output_image_path)
logging.info(f’Spectrogram saved to {output_image_path}‘)
plt.close()

# Хромаграмма
chromagram = librosa.feature.chroma_stft(y=y, sr=sr)
plt.figure(figsize=(12, 8))
librosa.display.specshow(chromagram, sr=sr, x_axis=‘time’, y_axis=‘chroma’)
plt.colorbar(format=’%+2.0’)
plt.title(‘Chromagram’)
plt.savefig(output_image_path.replace(‘_spectral_features.png’, ‘_chromagram.png’))
logging.info(f’Chromagram saved to {output_image_path.replace(“_spectral_features.png”, “_chromagram.png”)}‘)
plt.close()

# Спектральный центроид
spectral_centroids = librosa.feature.spectral_centroid(y=y, sr=sr)[0]
plt.figure(figsize=(12, 8))
plt.plot(spectral_centroids)
plt.xlabel(‘Time’)
plt.ylabel(‘Spectral Centroid (Hz)’)
plt.title(‘Spectral Centroid’)
plt.savefig(output_image_path.replace(’_spectral_features.png’, ‘_spectral_centroid.png’))
logging.info(f’Spectral Centroid saved to {output_image_path.replace(“_spectral_features.png”, “_spectral_centroid.png”)}‘)
plt.close()
except Exception as e:
logging.error(f"Error plotting spectral features: {e}")

def plot_harmonic_rhythm(tempo, beat_times, chords, output_image_path):
try:
import matplotlib.pyplot as plt
import librosa.display

# Темп
plt.figure(figsize=(12, 8))
plt.plot(beat_times, np.ones_like(beat_times), marker=’.‘, linestyle=‘None’)
plt.xlabel(‘Time’)
plt.ylabel(‘Beat’)
plt.title(f’Tempo: {tempo:.2f} BPM’)
plt.savefig(output_image_path.replace(‘_harmonic_rhythm.png’, ‘_tempo.png’))
logging.info(f’Tempo plot saved to {output_image_path.replace(“_harmonic_rhythm.png”, “_tempo.png”)}‘)
plt.close()

# Гармонический анализ
plt.figure(figsize=(12, 8))
librosa.display.specshow(chords, sr=1, x_axis=‘time’, y_axis=‘chroma’)
plt.colorbar()
plt.title(‘Harmonic Analysis’)
plt.savefig(output_image_path.replace(’_harmonic_rhythm.png’, ‘_harmonic_analysis.png’))
logging.info(f’Harmonic Analysis saved to {output_image_path.replace(“_harmonic_rhythm.png”, “_harmonic_analysis.png”)}‘)
plt.close()
except Exception as e:
logging.error(f"Error plotting harmonic rhythm: {e}")

def save_additional_data(file_path, cqt_db, frequency, confidence, transcription_wav2vec2, transcription_hubert, additional_features, instruments, genre_description, pitch_analysis, spectral_centroid, key, chroma_stft, output_dir):
try:
# Сохранение данных в файлы
np.save(os.path.join(output_dir, ‘cqt_db.npy’), cqt_db)
np.save(os.path.join(output_dir, ‘frequency.npy’), frequency)
np.save(os.path.join(output_dir, ‘confidence.npy’), confidence)
with open(os.path.join(output_dir, ‘transcription_wav2vec2.txt’), ‘w’) as f:
f.write(transcription_wav2vec2)
with open(os.path.join(output_dir, ‘transcription_hubert.txt’), ‘w’) as f:
f.write(transcription_hubert)
np.save(os.path.join(output_dir, ‘additional_features.npy’), additional_features)
with open(os.path.join(output_dir, ‘instruments.txt’), ‘w’) as f:
f.write(’, ‘.join(instruments))
with open(os.path.join(output_dir, ‘genre_description.txt’), ‘w’) as f:
f.write(genre_description)
np.save(os.path.join(output_dir, ‘pitch_analysis.npy’), pitch_analysis)
np.save(os.path.join(output_dir, ‘spectral_centroid.npy’), spectral_centroid)
with open(os.path.join(output_dir, ‘key.txt’), ‘w’) as f:
f.write(str(key))
np.save(os.path.join(output_dir, ‘chroma_stft.npy’), chroma_stft)

# Сохранение исходного аудиофайла
shutil.copy(file_path, os.path.join(output_dir, os.path.basename(file_path)))

logging.info(“Additional data saved successfully”)
except Exception as e:
logging.error(f"Error saving additional data: {e}")

# Ваш код здесь

# Путь к файлу аудио
file_path = r’C:\Users\ARTYOM\Documents\Python\Bring_it_up.mp3’

# Путь для сохранения MusicXML
output_path = r’C:\Users\ARTYOM\Documents\Python\Bring_it_up.musicxml’

# Путь для сохранения изображения партитуры
output_image_path = r’C:\Users\ARTYOM\Documents\Python\Bring_it_up.png’

# Время начала выполнения
start_time = time.time()

def save_musicxml_periodically(score, output_path, interval=60, use_music21=True):
“”“
Функция для периодического сохранения файла MusicXML.
“””
def save_periodically():
while True:
save_musicxml(score, output_path, use_music21)
time.sleep(interval)

thread = threading.Thread(target=save_periodically)
thread.daemon = True
thread.start()

async def main_async(file_path, output_path, output_image_path):
# Инициализация конвейера и запуск обработки
audio_pipeline = AudioPipeline(config)
results = await audio_pipeline.run_pipeline(file_path)

if results:
# Обработка результатов
(cqt_db, frequency, confidence, transcription_wav2vec2, transcription_hubert, additional_features,
sources_openunmix, sources_demucs_v3, vocals_path, accompaniment_path, pitch_analysis, normalized_audio, optimized_audio) = results

logging.info(f"Processing results: {cqt_db.shape}, {len(frequency)}, {transcription_wav2vec2}, {transcription_hubert}“)

# Гармонический анализ и ритм
y, sr = librosa.load(file_path, sr=None)
tempo, beat_frames, chords, beat_times = harmonic_rhythm_analysis(y, sr)
if tempo is None or beat_frames is None or chords is None or beat_times is None:
logging.error(“Error in harmonic rhythm analysis”)
return

# Генерация текстовых описаний музыкальных фрагментов с использованием ChatGPT (GPT-4)
text_descriptions = generate_text_descriptions(additional_features)
if text_descriptions is None:
logging.error(“Error generating text descriptions”)
return

# Распознавание музыкальных инструментов
instruments = audio_pipeline.recognize_instruments(y, sr)
if instruments is None:
logging.error(“Error recognizing instruments”)
return

# Распознавание жанра музыки
genre_description = audio_pipeline.recognize_genre(y, sr)
if genre_description is None:
logging.error(“Error recognizing genre”)
return

# Спектральный центроид
spectral_centroid = compute_spectral_centroid(y, sr)
if spectral_centroid is None:
logging.error(“Error computing spectral centroid”)
return

# Тональность и тональная стабильность
key, chroma_stft = compute_tonality_and_stability(y, sr)
if key is None or chroma_stft is None:
logging.error(“Error computing tonality and stability”)
return

# Создание MusicXML из результатов
# Используем обе библиотеки для преобразования
score_music21 = create_musicxml(frequency, beat_times, range(len(beat_times)), range(len(beat_times)),
range(len(beat_times)), text_descriptions, instruments, use_music21=True, use_lxml=False)
score_lxml = create_musicxml(frequency, beat_times, range(len(beat_times)), range(len(beat_times)),
range(len(beat_times)), text_descriptions, instruments, use_music21=False, use_lxml=True)

if score_music21 is None and score_lxml is None:
logging.error(“Error creating MusicXML”)
return

# Логика добавления или удаления нот с использованием GPT-4 (ChatGPT)
score_music21 = modify_notes_with_gpt4(score_music21, action=“add”, instrument=“piano”)

# Сохранение и визуализация партитуры в формате MusicXML и изображения
output_path_music21 = output_path.replace(‘.xml’, ‘_music21.xml’)
output_path_lxml = output_path.replace(‘.xml’, ‘_lxml.xml’)

save_musicxml(score_music21, output_path_music21, use_music21=True)
save_musicxml(score_lxml, output_path_lxml, use_music21=False)

visualize_score(score_music21, output_image_path)

# Визуализация спектрограммы аудиофайла
spectrogram_image_path = output_image_path.replace(”.png", “_spectrogram.png”)
y, sr = await preprocess_audio(file_path)
plot_spectrogram(y, sr, spectrogram_image_path)

# Визуализация хромограммы аудиофайла
chromagram_image_path = output_image_path.replace(“.png”, “_chromagram.png”)
plot_chromagram(y, sr, chromagram_image_path)

# Демонстрация воспроизведения аудио
audio_pipeline.play_audio(normalized_audio)
audio_pipeline.play_audio(optimized_audio)

# Окончательное описание всех источников
full_description = {
“Transcription (Wav2Vec2)”: transcription_wav2vec2,
“Transcription (HuBERT)”: transcription_hubert,
“Genre Description (GPT-4)”: genre_description,
“Instrument Recognition (YAMNet)”: instruments,
“Additional Features (Essentia)”: additional_features,
“Pitch Analysis (Parselmouth)”: pitch_analysis,
“Spectral Centroid”: spectral_centroid,
“Key (Tuning)”: key,
“Chroma (Tonality Stability)”: chroma_stft
}

logging.info(“Full Description:”)
for key, value in full_description.items():
logging.info(f"{key}: {value}“)

# Сохранение оптимизированного аудио
optimized_audio_path = os.path.join(config[“output_dir”], “optimized_audio.wav”)
sf.write(optimized_audio_path, optimized_audio, sr)
logging.info(f"Optimized audio saved to: {optimized_audio_path}”)

# Сохранение разделенных источников
if vocals_path and accompaniment_path:
logging.info(f"Vocals saved to: {vocals_path}“)
logging.info(f"Accompaniment saved to: {accompaniment_path}”)

# Сохранение дополнительных визуализаций
plot_spectral_features(y, sr, output_image_path.replace(“.png”, “_spectral_features.png”))
plot_harmonic_rhythm(tempo, beat_times, chords, output_image_path.replace(“.png”, “_harmonic_rhythm.png”))

# Сохранение дополнительных данных
save_additional_data(
file_path,
cqt_db, frequency, confidence,
transcription_wav2vec2, transcription_hubert,
additional_features, instruments, genre_description,
pitch_analysis, spectral_centroid, key, chroma_stft,
output_dir=config[“output_dir”]
)

# Логирование времени выполнения
start_time = time.time()
asyncio.run(main_async(file_path, output_path, output_image_path))
logging.info(f"Execution time: {time.time() - start_time} seconds")

# Запуск потоковой обработки и воспроизведения аудио в реальном времени
async def stream_audio(file_path):
try:
await play_and_process_audio(file_path)
except Exception as e:
logging.error(f"Error in stream_audio: {e}“)

# Основной запуск
if name == “main”:
# Запуск основного асинхронного процесса
asyncio.run(main_async(file_path, output_path, output_image_path))

# Запуск потоковой обработки и воспроизведения аудио в реальном времени
asyncio.run(stream_audio(file_path))

def plot_spectral_features(y, sr, output_image_path):
try:
import matplotlib.pyplot as plt
import librosa.display

# Спектрограмма
D = librosa.amplitude_to_db(np.abs(librosa.stft(y)), ref=np.max)
plt.figure(figsize=(12, 8))
librosa.display.specshow(D, sr=sr, x_axis=‘time’, y_axis=‘log’)
plt.colorbar(format=‘%+2.0f dB’)
plt.title(‘Spectrogram’)
plt.savefig(output_image_path)
logging.info(f’Spectrogram saved to {output_image_path}‘)
plt.close()

# Хромаграмма
chromagram = librosa.feature.chroma_stft(y=y, sr=sr)
plt.figure(figsize=(12, 8))
librosa.display.specshow(chromagram, sr=sr, x_axis=‘time’, y_axis=‘chroma’)
plt.colorbar(format=’%+2.0’)
plt.title(‘Chromagram’)
plt.savefig(output_image_path.replace(‘_spectral_features.png’, ‘_chromagram.png’))
logging.info(f’Chromagram saved to {output_image_path.replace(”_spectral_features.png", “_chromagram.png”)}‘)
plt.close()

# Спектральный центроид
spectral_centroids = librosa.feature.spectral_centroid(y=y, sr=sr)[0]
plt.figure(figsize=(12, 8))
plt.plot(spectral_centroids)
plt.xlabel(‘Time’)
plt.ylabel(‘Spectral Centroid (Hz)’)
plt.title(‘Spectral Centroid’)
plt.savefig(output_image_path.replace(’_spectral_features.png’, ‘_spectral_centroid.png’))
logging.info(f’Spectral Centroid saved to {output_image_path.replace(“_spectral_features.png”, “_spectral_centroid.png”)}‘)
plt.close()
except Exception as e:
logging.error(f"Error plotting spectral features: {e}")

def plot_harmonic_rhythm(tempo, beat_times, chords, output_image_path):
try:
import matplotlib.pyplot as plt
import librosa.display

# Темп
plt.figure(figsize=(12, 8))
plt.plot(beat_times, np.ones_like(beat_times), marker=’.‘, linestyle=‘None’)
plt.xlabel(‘Time’)
plt.ylabel(‘Beat’)
plt.title(f’Tempo: {tempo:.2f} BPM’)
plt.savefig(output_image_path.replace(‘_harmonic_rhythm.png’, ‘_tempo.png’))
logging.info(f’Tempo plot saved to {output_image_path.replace(“_harmonic_rhythm.png”, “_tempo.png”)}‘)
plt.close()

# Гармонический анализ
plt.figure(figsize=(12, 8))
librosa.display.specshow(chords, sr=1, x_axis=‘time’, y_axis=‘chroma’)
plt.colorbar()
plt.title(‘Harmonic Analysis’)
plt.savefig(output_image_path.replace(’_harmonic_rhythm.png’, ‘_harmonic_analysis.png’))
logging.info(f’Harmonic Analysis saved to {output_image_path.replace(“_harmonic_rhythm.png”, “_harmonic_analysis.png”)}‘)
plt.close()
except Exception as e:
logging.error(f"Error plotting harmonic rhythm: {e}")

def save_additional_data(file_path, cqt_db, frequency, confidence, transcription_wav2vec2, transcription_hubert, additional_features, instruments, genre_description, pitch_analysis, spectral_centroid, key, chroma_stft, output_dir):
try:
# Сохранение данных в файлы
np.save(os.path.join(output_dir, ‘cqt_db.npy’), cqt_db)
np.save(os.path.join(output_dir, ‘frequency.npy’), frequency)
np.save(os.path.join(output_dir, ‘confidence.npy’), confidence)
with open(os.path.join(output_dir, ‘transcription_wav2vec2.txt’), ‘w’) as f:
f.write(transcription_wav2vec2)
with open(os.path.join(output_dir, ‘transcription_hubert.txt’), ‘w’) as f:
f.write(transcription_hubert)
np.save(os.path.join(output_dir, ‘additional_features.npy’), additional_features)
with open(os.path.join(output_dir, ‘instruments.txt’), ‘w’) as f:
f.write(’, ‘.join(instruments))
with open(os.path.join(output_dir, ‘genre_description.txt’), ‘w’) as f:
f.write(genre_description)
np.save(os.path.join(output_dir, ‘pitch_analysis.npy’), pitch_analysis)
np.save(os.path.join(output_dir, ‘spectral_centroid.npy’), spectral_centroid)
with open(os.path.join(output_dir, ‘key.txt’), ‘w’) as f:
f.write(str(key))
np.save(os.path.join(output_dir, ‘chroma_stft.npy’), chroma_stft)

# Сохранение исходного аудиофайла
shutil.copy(file_path, os.path.join(output_dir, os.path.basename(file_path)))

logging.info(“Additional data saved successfully”)
except Exception as e:
logging.error(f"Error saving additional data: {e}")

# Ваш код здесь

# Путь к файлу аудио
file_path = r’C:\Users\ARTYOM\Documents\Python\Bring_it_up.mp3’

# Путь для сохранения MusicXML
output_path = r’C:\Users\ARTYOM\Documents\Python\Bring_it_up.musicxml’

# Путь для сохранения изображения партитуры
output_image_path = r’C:\Users\ARTYOM\Documents\Python\Bring_it_up.png’

# Время начала выполнения
start_time = time.time()

def save_musicxml_periodically(score, output_path, interval=60, use_music21=True):
“”“
Функция для периодического сохранения файла MusicXML.
“””
def save_periodically():
while True:
save_musicxml(score, output_path, use_music21)
time.sleep(interval)

thread = threading.Thread(target=save_periodically)
thread.daemon = True
thread.start()

async def main_async(file_path, output_path, output_image_path):
# Инициализация конвейера и запуск обработки
audio_pipeline = AudioPipeline(config)
results = await audio_pipeline.run_pipeline(file_path)

if results:
# Обработка результатов
(cqt_db, frequency, confidence, transcription_wav2vec2, transcription_hubert, additional_features,
sources_openunmix, sources_demucs_v3, vocals_path, accompaniment_path, pitch_analysis, normalized_audio, optimized_audio) = results

logging.info(f"Processing results: {cqt_db.shape}, {len(frequency)}, {transcription_wav2vec2}, {transcription_hubert}“)

# Гармонический анализ и ритм
y, sr = librosa.load(file_path, sr=None)
tempo, beat_frames, chords, beat_times = harmonic_rhythm_analysis(y, sr)
if tempo is None or beat_frames is None or chords is None or beat_times is None:
logging.error(“Error in harmonic rhythm analysis”)
return

# Генерация текстовых описаний музыкальных фрагментов с использованием ChatGPT (GPT-4)
text_descriptions = generate_text_descriptions(additional_features)
if text_descriptions is None:
logging.error(“Error generating text descriptions”)
return

# Распознавание музыкальных инструментов
instruments = audio_pipeline.recognize_instruments(y, sr)
if instruments is None:
logging.error(“Error recognizing instruments”)
return

# Распознавание жанра музыки
genre_description = audio_pipeline.recognize_genre(y, sr)
if genre_description is None:
logging.error(“Error recognizing genre”)
return

# Спектральный центроид
spectral_centroid = compute_spectral_centroid(y, sr)
if spectral_centroid is None:
logging.error(“Error computing spectral centroid”)
return

# Тональность и тональная стабильность
key, chroma_stft = compute_tonality_and_stability(y, sr)
if key is None or chroma_stft is None:
logging.error(“Error computing tonality and stability”)
return

# Создание MusicXML из результатов
# Используем обе библиотеки для преобразования
score_music21 = create_musicxml(frequency, beat_times, range(len(beat_times)), range(len(beat_times)),
range(len(beat_times)), text_descriptions, instruments, use_music21=True, use_lxml=False)
score_lxml = create_musicxml(frequency, beat_times, range(len(beat_times)), range(len(beat_times)),
range(len(beat_times)), text_descriptions, instruments, use_music21=False, use_lxml=True)

if score_music21 is None and score_lxml is None:
logging.error(“Error creating MusicXML”)
return

# Логика добавления или удаления нот с использованием GPT-4 (ChatGPT)
score_music21 = modify_notes_with_gpt4(score_music21, action=“add”, instrument=“piano”)

# Сохранение и визуализация партитуры в формате MusicXML и изображения
output_path_music21 = output_path.replace(‘.xml’, ‘_music21.xml’)
output_path_lxml = output_path.replace(‘.xml’, ‘_lxml.xml’)

save_musicxml(score_music21, output_path_music21, use_music21=True)
save_musicxml(score_lxml, output_path_lxml, use_music21=False)

visualize_score(score_music21, output_image_path)

# Визуализация спектрограммы аудиофайла
spectrogram_image_path = output_image_path.replace(”.png", “_spectrogram.png”)
y, sr = await preprocess_audio(file_path)
plot_spectrogram(y, sr, spectrogram_image_path)

# Визуализация хромограммы аудиофайла
chromagram_image_path = output_image_path.replace(“.png”, “_chromagram.png”)
plot_chromagram(y, sr, chromagram_image_path)

# Демонстрация воспроизведения аудио
audio_pipeline.play_audio(normalized_audio)
audio_pipeline.play_audio(optimized_audio)

# Окончательное описание всех источников
full_description = {
“Transcription (Wav2Vec2)”: transcription_wav2vec2,
“Transcription (HuBERT)”: transcription_hubert,
“Genre Description (GPT-4)”: genre_description,
“Instrument Recognition (YAMNet)”: instruments,
“Additional Features (Essentia)”: additional_features,
“Pitch Analysis (Parselmouth)”: pitch_analysis,
“Spectral Centroid”: spectral_centroid,
“Key (Tuning)”: key,
“Chroma (Tonality Stability)”: chroma_stft
}

logging.info(“Full Description:”)
for key, value in full_description.items():
logging.info(f"{key}: {value}“)

# Сохранение оптимизированного аудио
optimized_audio_path = os.path.join(config[“output_dir”], “optimized_audio.wav”)
sf.write(optimized_audio_path, optimized_audio, sr)
logging.info(f"Optimized audio saved to: {optimized_audio_path}”)

# Сохранение разделенных источников
if vocals_path and accompaniment_path:
logging.info(f"Vocals saved to: {vocals_path}“)
logging.info(f"Accompaniment saved to: {accompaniment_path}”)

# Сохранение дополнительных визуализаций
plot_spectral_features(y, sr, output_image_path.replace(“.png”, “_spectral_features.png”))
plot_harmonic_rhythm(tempo, beat_times, chords, output_image_path.replace(“.png”, “_harmonic_rhythm.png”))

# Сохранение дополнительных данных
save_additional_data(
file_path,
cqt_db, frequency, confidence,
transcription_wav2vec2, transcription_hubert,
additional_features, instruments, genre_description,
pitch_analysis, spectral_centroid, key, chroma_stft,
output_dir=config[“output_dir”]
)

# Логирование времени выполнения
start_time = time.time()
asyncio.run(main_async(file_path, output_path, output_image_path))
logging.info(f"Execution time: {time.time() - start_time} seconds")

# Запуск потоковой обработки и воспроизведения аудио в реальном времени
async def stream_audio(file_path):
try:
await play_and_process_audio(file_path)
except Exception as e:
logging.error(f"Error in stream_audio: {e}")