feat: thêm soundfont và VSTi cho MIDI
This commit is contained in:
+71
-25
@@ -2,24 +2,13 @@ import os
|
||||
import numpy as np
|
||||
import soundfile as sf
|
||||
from app.config import settings
|
||||
from app.core.vst_engine import render_midi_events_to_audio
|
||||
from app.core.vst_engine import (
|
||||
render_midi_events_to_audio,
|
||||
PluginManager,
|
||||
HAS_PEDALBOARD,
|
||||
HAS_PYFLUIDSYNTH,
|
||||
)
|
||||
|
||||
import subprocess
|
||||
import sys
|
||||
|
||||
def check_pedalboard_safe():
|
||||
try:
|
||||
res = subprocess.run(
|
||||
[sys.executable, "-c", "import pedalboard"],
|
||||
stdout=subprocess.DEVNULL,
|
||||
stderr=subprocess.DEVNULL,
|
||||
timeout=2.0
|
||||
)
|
||||
return res.returncode == 0
|
||||
except Exception:
|
||||
return False
|
||||
|
||||
HAS_PEDALBOARD = check_pedalboard_safe()
|
||||
if HAS_PEDALBOARD:
|
||||
try:
|
||||
from pedalboard import Pedalboard, Gain
|
||||
@@ -130,14 +119,71 @@ class PythonRenderEngine:
|
||||
|
||||
if midi_events:
|
||||
try:
|
||||
# Synthesize MIDI track notes
|
||||
synth_buffer = render_midi_events_to_audio(
|
||||
midi_events=midi_events,
|
||||
sr=self.sample_rate,
|
||||
bpm=bpm,
|
||||
instrument='synth'
|
||||
)
|
||||
# Add to track buffer
|
||||
instrument_id = track.get("instrument", "")
|
||||
plugin_mgr = PluginManager()
|
||||
vst = plugin_mgr.load_vst(instrument_id) if instrument_id else None
|
||||
|
||||
if vst and HAS_PEDALBOARD:
|
||||
from pedalboard import Pedalboard
|
||||
# Convert MIDI events with precise sample offset
|
||||
midi_messages = PluginManager.midi_events_to_messages(
|
||||
midi_events, bpm, self.sample_rate
|
||||
)
|
||||
total_needed = 0
|
||||
for ev in midi_events:
|
||||
end_sec = (ev.get("start_beat", 0) + ev.get("duration_beats", 1)) * (60.0 / bpm)
|
||||
dur_samples = int(end_sec * self.sample_rate)
|
||||
if dur_samples > total_needed:
|
||||
total_needed = dur_samples
|
||||
total_needed = max(total_needed, 1024)
|
||||
silent = np.zeros((2, total_needed), dtype=np.float32)
|
||||
board = Pedalboard([vst])
|
||||
synth_buffer = board(silent, sample_rate=self.sample_rate, midi_messages=midi_messages)
|
||||
elif instrument_id and instrument_id.startswith("sf_"):
|
||||
sf_path = os.path.join(
|
||||
os.path.dirname(os.path.dirname(__file__)), "static", "soundfonts",
|
||||
instrument_id.replace("sf_", "") + ".sf2"
|
||||
)
|
||||
if os.path.exists(sf_path) and HAS_PYFLUIDSYNTH:
|
||||
import fluidsynth
|
||||
fl = fluidsynth.FluidSynth(sample_rate=self.sample_rate, gain=0.5)
|
||||
fid = fl.sfload(sf_path)
|
||||
fl.program_select(0, fid, 0, 0)
|
||||
beat_sec = 60.0 / bpm
|
||||
total_sec = 0
|
||||
for ev in midi_events:
|
||||
end_sec = (ev.get("start_beat", 0) + ev.get("duration_beats", 1)) * beat_sec
|
||||
if end_sec > total_sec:
|
||||
total_sec = end_sec
|
||||
total_samples = int((total_sec + 1.0) * self.sample_rate)
|
||||
midi_data = np.zeros((2, total_samples), dtype=np.float32)
|
||||
for ev in midi_events:
|
||||
note = ev.get("note", 60)
|
||||
velocity = ev.get("velocity", 100)
|
||||
start_beat = ev.get("start_beat", 0.0)
|
||||
dur_beats = ev.get("duration_beats", 1.0)
|
||||
start_sec = start_beat * beat_sec
|
||||
dur_sec = dur_beats * beat_sec
|
||||
fl.noteon(0, note, velocity)
|
||||
start_s = int(start_sec * self.sample_rate)
|
||||
dur_s = int(dur_sec * self.sample_rate)
|
||||
block = fl.get_samples(int(dur_s)) if hasattr(fl, 'get_samples') else np.zeros((2, dur_s), dtype=np.float32)
|
||||
fl.noteoff(0, note)
|
||||
if block.shape[1] > 0:
|
||||
end_s = min(start_s + block.shape[1], total_samples)
|
||||
actual = end_s - start_s
|
||||
if actual > 0:
|
||||
midi_data[:, start_s:end_s] += block[:, :actual]
|
||||
synth_buffer = midi_data
|
||||
fl.delete()
|
||||
else:
|
||||
synth_buffer = render_midi_events_to_audio(
|
||||
midi_events=midi_events, sr=self.sample_rate, bpm=bpm, instrument='synth'
|
||||
)
|
||||
else:
|
||||
synth_buffer = render_midi_events_to_audio(
|
||||
midi_events=midi_events, sr=self.sample_rate, bpm=bpm, instrument='synth'
|
||||
)
|
||||
actual_len = min(synth_buffer.shape[1], total_samples)
|
||||
track_buffer[:, :actual_len] += synth_buffer[:, :actual_len]
|
||||
except Exception as e:
|
||||
|
||||
+131
-31
@@ -1,77 +1,177 @@
|
||||
# SonicForge Studio VST / VSTi Engine Service (22_CLIENT_DESK.md §3)
|
||||
# SonicForge Studio VST / VSTi Engine Service
|
||||
import os
|
||||
import numpy as np
|
||||
|
||||
def midi_note_to_freq(note_number: int) -> float:
|
||||
"""Quy đổi số nốt MIDI (0 - 127) sang tần số Hertz (Hz)."""
|
||||
return 440.0 * (2.0 ** ((note_number - 69) / 12.0))
|
||||
|
||||
def render_midi_events_to_audio(midi_events: list, sr: int = 44100, bpm: float = 120.0, instrument: str = 'synth') -> np.ndarray:
|
||||
"""
|
||||
Tổng hợp mảng âm thanh NumPy Stereo từ sự kiện MIDI Piano Roll (22_CLIENT_DESK.md §3.1 & §3.2).
|
||||
|
||||
Args:
|
||||
midi_events: Danh sách nốt MIDI [{"note": 60, "start_beat": 0, "duration_beats": 1, "velocity": 100}, ...]
|
||||
sr: Tần số lấy mẫu (Sample Rate)
|
||||
bpm: Nhịp BPM của dự án
|
||||
instrument: Loại nhạc cụ tổng hợp
|
||||
|
||||
Returns:
|
||||
np.ndarray: Mảng 2D Stereo Float32 [2, num_samples]
|
||||
"""
|
||||
beat_duration_sec = 60.0 / max(30.0, bpm)
|
||||
max_duration_sec = 2.0
|
||||
|
||||
for event in midi_events:
|
||||
start_beat = event.get('start_beat', 0.0)
|
||||
dur_beats = event.get('duration_beats', 1.0)
|
||||
end_sec = (start_beat + dur_beats) * beat_duration_sec
|
||||
if end_sec > max_duration_sec:
|
||||
max_duration_sec = end_sec
|
||||
|
||||
total_samples = int((max_duration_sec + 0.5) * sr)
|
||||
out_l = np.zeros(total_samples, dtype=np.float32)
|
||||
out_r = np.zeros(total_samples, dtype=np.float32)
|
||||
|
||||
for event in midi_events:
|
||||
note = event.get('note', 60)
|
||||
velocity = event.get('velocity', 100) / 127.0
|
||||
start_beat = event.get('start_beat', 0.0)
|
||||
dur_beats = event.get('duration_beats', 1.0)
|
||||
|
||||
start_sample = int(start_beat * beat_duration_sec * sr)
|
||||
dur_samples = int(dur_beats * beat_duration_sec * sr)
|
||||
end_sample = min(total_samples, start_sample + dur_samples)
|
||||
actual_len = end_sample - start_sample
|
||||
|
||||
if actual_len <= 0 or start_sample >= total_samples:
|
||||
continue
|
||||
|
||||
freq = midi_note_to_freq(note)
|
||||
t = np.arange(actual_len) / float(sr)
|
||||
|
||||
# Synth tone + fundamental harmonics
|
||||
tone = 0.6 * np.sin(2 * np.pi * freq * t) + 0.3 * np.sin(2 * np.pi * freq * 2 * t) + 0.1 * np.sin(2 * np.pi * freq * 3 * t)
|
||||
|
||||
# ADSR Envelope
|
||||
attack = min(int(0.01 * sr), actual_len // 4)
|
||||
release = min(int(0.05 * sr), actual_len // 4)
|
||||
sustain_len = actual_len - attack - release
|
||||
|
||||
env = np.ones(actual_len, dtype=np.float32)
|
||||
if attack > 0:
|
||||
env[:attack] = np.linspace(0.0, 1.0, attack)
|
||||
if release > 0:
|
||||
env[-release:] = np.linspace(1.0, 0.0, release)
|
||||
|
||||
signal = tone * env * velocity
|
||||
|
||||
out_l[start_sample:end_sample] += signal
|
||||
out_r[start_sample:end_sample] += signal
|
||||
|
||||
# Clamping normalization to prevent clipping
|
||||
max_peak = max(np.max(np.abs(out_l)), np.max(np.abs(out_r)))
|
||||
if max_peak > 1.0:
|
||||
out_l /= max_peak
|
||||
out_r /= max_peak
|
||||
|
||||
return np.vstack([out_l, out_r])
|
||||
|
||||
def check_pedalboard_safe():
|
||||
import subprocess, sys
|
||||
try:
|
||||
res = subprocess.run(
|
||||
[sys.executable, "-c", "import pedalboard"],
|
||||
stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, timeout=2.0
|
||||
)
|
||||
return res.returncode == 0
|
||||
except Exception:
|
||||
return False
|
||||
|
||||
def check_pyfluidsynth_safe():
|
||||
import subprocess, sys
|
||||
try:
|
||||
res = subprocess.run(
|
||||
[sys.executable, "-c", "import fluidsynth"],
|
||||
stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, timeout=2.0
|
||||
)
|
||||
return res.returncode == 0
|
||||
except Exception:
|
||||
return False
|
||||
|
||||
HAS_PEDALBOARD = check_pedalboard_safe()
|
||||
HAS_PYFLUIDSYNTH = check_pyfluidsynth_safe()
|
||||
|
||||
if HAS_PEDALBOARD:
|
||||
try:
|
||||
from pedalboard import VST3Plugin, Pedalboard, Gain, MidiMessage
|
||||
except Exception:
|
||||
HAS_PEDALBOARD = False
|
||||
|
||||
if HAS_PYFLUIDSYNTH:
|
||||
try:
|
||||
import fluidsynth
|
||||
except Exception:
|
||||
HAS_PYFLUIDSYNTH = False
|
||||
|
||||
|
||||
class PluginManager:
|
||||
def __init__(self, vst_dir="/opt/daw_engine/vst3", sf_dir="/opt/daw_engine/soundfonts"):
|
||||
self.vst_dir = vst_dir
|
||||
self.sf_dir = sf_dir
|
||||
|
||||
def _scan_plugins(self) -> dict:
|
||||
plugins = {}
|
||||
if not os.path.isdir(self.vst_dir):
|
||||
return plugins
|
||||
for root, dirs, files in os.walk(self.vst_dir):
|
||||
for file in files:
|
||||
if file.endswith(".vst3") or file.endswith(".so"):
|
||||
plugin_path = os.path.join(root, file)
|
||||
plugin_name = os.path.splitext(file)[0]
|
||||
plugins[plugin_name] = plugin_path
|
||||
return plugins
|
||||
|
||||
def _scan_soundfonts(self) -> list:
|
||||
sfonts = []
|
||||
if not os.path.isdir(self.sf_dir):
|
||||
return sfonts
|
||||
for f in os.listdir(self.sf_dir):
|
||||
if f.endswith(".sf2") or f.endswith(".sf3"):
|
||||
sfonts.append({"id": os.path.splitext(f)[0], "name": f, "file": f})
|
||||
return sfonts
|
||||
|
||||
def load_vst(self, plugin_name: str, preset_data: dict = None):
|
||||
if not HAS_PEDALBOARD:
|
||||
return None
|
||||
plugins = self._scan_plugins()
|
||||
if plugin_name not in plugins:
|
||||
return None
|
||||
path = plugins[plugin_name]
|
||||
vst = VST3Plugin(path)
|
||||
if preset_data:
|
||||
for k, v in preset_data.items():
|
||||
try:
|
||||
setattr(vst, k, v)
|
||||
except Exception:
|
||||
pass
|
||||
return vst
|
||||
|
||||
def load_soundfont(self, path: str):
|
||||
if not HAS_PYFLUIDSYNTH:
|
||||
return None
|
||||
try:
|
||||
fl = fluidsynth.FluidSynth(sample_rate=44100, gain=0.5)
|
||||
font_id = fl.sfload(path)
|
||||
fl.program_select(0, font_id, 0, 0)
|
||||
return fl
|
||||
except Exception:
|
||||
return None
|
||||
|
||||
def list_available(self) -> dict:
|
||||
return {
|
||||
"vst_instruments": [
|
||||
{"id": k, "name": k, "type": "VST3", "has_native_support": HAS_PEDALBOARD}
|
||||
for k in self._scan_plugins().keys()
|
||||
],
|
||||
"soundfonts": self._scan_soundfonts()
|
||||
}
|
||||
|
||||
@staticmethod
|
||||
def midi_events_to_messages(midi_events: list, bpm: float, sr: int) -> list:
|
||||
if not HAS_PEDALBOARD:
|
||||
return []
|
||||
beat_duration_sec = 60.0 / max(30.0, bpm)
|
||||
messages = []
|
||||
for ev in midi_events:
|
||||
note = ev.get("note", 60)
|
||||
velocity = ev.get("velocity", 100)
|
||||
start_beat = ev.get("start_beat", 0.0)
|
||||
dur_beats = ev.get("duration_beats", 1.0)
|
||||
start_sec = start_beat * beat_duration_sec
|
||||
dur_sec = dur_beats * beat_duration_sec
|
||||
sample_offset = int(start_sec * sr)
|
||||
end_sample_offset = int((start_sec + dur_sec) * sr)
|
||||
messages.append(MidiMessage(note_on=note, velocity=velocity, sample_offset=sample_offset))
|
||||
messages.append(MidiMessage(note_off=note, velocity=0, sample_offset=end_sample_offset))
|
||||
return messages
|
||||
|
||||
@staticmethod
|
||||
def validate_sf2_header(data: bytes) -> bool:
|
||||
if len(data) < 12:
|
||||
return False
|
||||
if data[0:4] != b'RIFF':
|
||||
return False
|
||||
if data[8:12] != b'sfbk':
|
||||
return False
|
||||
return True
|
||||
|
||||
Reference in New Issue
Block a user