3dtours
89c7237379
- SoundFontInspector (sf2utils) scans .sf2, generates full/condensed catalog - GET /api/v1/plugins/soundfonts/catalog with lazy init + cache invalidation - AI tool generate_multitrack_midi now requires soundfont_id/bank/program - Condensed catalog auto-injected into AI system prompt with bank rules - Server render: FluidSynth program_select uses bank/program + channel routing (drums→ch9) - VST3 pedalboard path inserts CC0 bank select + program change before notes - DecentSamplerManager loads .dspreset with CWD fix for relative sample paths - Pianobook render branch in render_engine.py - Client SonicSF: controllerChange, programChange, applyAITrackInstrument - Post-AI track creation applies instrument via applyAITrackInstrument - Background cache rescan on .sf2 upload, frontend re-fetches catalog - libcurl4 + VST3 dirs in Dockerfile
370 lines
20 KiB
Python
370 lines
20 KiB
Python
import os
|
|
import numpy as np
|
|
import soundfile as sf
|
|
import scipy.signal as signal
|
|
from app.config import settings
|
|
from app.core.vst_engine import (
|
|
render_midi_events_to_audio,
|
|
PluginManager,
|
|
DecentSamplerManager,
|
|
HAS_PEDALBOARD,
|
|
HAS_PYFLUIDSYNTH,
|
|
)
|
|
|
|
if HAS_PEDALBOARD:
|
|
try:
|
|
from pedalboard import Pedalboard, Gain, Chorus, Reverb
|
|
except Exception:
|
|
HAS_PEDALBOARD = False
|
|
|
|
|
|
|
|
class PythonRenderEngine:
|
|
def __init__(self, sample_rate=44100):
|
|
self.sample_rate = sample_rate
|
|
|
|
def bars_to_samples(self, bars: float, bpm: float, time_sig_num: int) -> int:
|
|
seconds_per_beat = 60.0 / max(20.0, bpm)
|
|
seconds_per_bar = seconds_per_beat * time_sig_num
|
|
return int(bars * seconds_per_bar * self.sample_rate)
|
|
|
|
def resolve_file_path(self, url_or_id: str) -> str:
|
|
if not url_or_id:
|
|
return ""
|
|
base = os.path.basename(url_or_id)
|
|
# Check uploads directory
|
|
p_uploads = os.path.join(settings.UPLOADS_DIR, base)
|
|
if os.path.exists(p_uploads):
|
|
return p_uploads
|
|
# Check processed directory
|
|
p_processed = os.path.join(settings.PROCESSED_DIR, base)
|
|
if os.path.exists(p_processed):
|
|
return p_processed
|
|
# Check general storage directory
|
|
p_storage = os.path.join(settings.STORAGE_DIR, base)
|
|
if os.path.exists(p_storage):
|
|
return p_storage
|
|
# Direct check
|
|
if os.path.exists(url_or_id):
|
|
return url_or_id
|
|
return url_or_id
|
|
|
|
def render_session_container(self, session: dict, section_store: dict, bpm: float, time_sig_num: int, total_samples: int) -> np.ndarray:
|
|
session_buffer = np.zeros((2, total_samples), dtype=np.float32)
|
|
|
|
_channel_counter = 0
|
|
|
|
for track in session.get("tracks", []):
|
|
track_type = track.get("type", "AUDIO")
|
|
track_buffer = np.zeros((2, total_samples), dtype=np.float32)
|
|
|
|
soundfont_bank = track.get("soundfont_bank", 0)
|
|
soundfont_program = track.get("soundfont_program", 0)
|
|
is_percussion = track.get("is_percussion", False) or (soundfont_bank == 128)
|
|
midi_channel = 9 if is_percussion else (_channel_counter % 9)
|
|
if not is_percussion:
|
|
_channel_counter += 1
|
|
|
|
for item in track.get("items", []):
|
|
start_sample = self.bars_to_samples(item["start_bar"], bpm, time_sig_num)
|
|
dur_samples = self.bars_to_samples(item["duration_bars"], bpm, time_sig_num)
|
|
offset_sample = self.bars_to_samples(item["clip_start_offset_bars"], bpm, time_sig_num)
|
|
|
|
item_type = item.get("type")
|
|
if item_type == "AUDIO_ITEM":
|
|
source_data = item.get("source_data", {})
|
|
audio_url = source_data.get("audio_file_url", "")
|
|
resolved_path = self.resolve_file_path(audio_url)
|
|
|
|
if resolved_path and os.path.exists(resolved_path):
|
|
try:
|
|
audio_data, sr = sf.read(resolved_path, dtype='float32')
|
|
if sr != self.sample_rate:
|
|
# Resampling fallback if simple, otherwise skip
|
|
pass
|
|
|
|
# Handle channel mapping (Mono/Stereo)
|
|
if len(audio_data.shape) == 1:
|
|
audio_data = np.vstack([audio_data, audio_data])
|
|
else:
|
|
audio_data = audio_data.T # Shape: (channels, samples)
|
|
|
|
# Trim source offset & duration
|
|
src_len = audio_data.shape[1]
|
|
if offset_sample < src_len:
|
|
actual_dur = min(dur_samples, src_len - offset_sample)
|
|
sliced_audio = audio_data[:, offset_sample : offset_sample + actual_dur]
|
|
|
|
# Apply gain
|
|
gain_val = source_data.get("gain", 1.0)
|
|
sliced_audio = sliced_audio * gain_val
|
|
|
|
# Write to track buffer with boundaries
|
|
write_end = min(start_sample + sliced_audio.shape[1], total_samples)
|
|
actual_len = write_end - start_sample
|
|
if actual_len > 0:
|
|
track_buffer[:, start_sample:write_end] += sliced_audio[:, :actual_len]
|
|
except Exception as e:
|
|
print(f"[RenderEngine] Error reading audio file {resolved_path}: {e}")
|
|
|
|
elif item_type == "MIDI_ITEM":
|
|
source_data = item.get("source_data", {})
|
|
notes = source_data.get("notes", [])
|
|
|
|
# Convert to midi events required by vst_engine
|
|
midi_events = []
|
|
for note in notes:
|
|
note_start_bar = note["start_beat"] / time_sig_num
|
|
# Filter notes within the non-destructive visible window
|
|
offset_bar = item["clip_start_offset_bars"]
|
|
dur_bar = item["duration_bars"]
|
|
if note_start_bar >= offset_bar and note_start_bar < (offset_bar + dur_bar):
|
|
rel_bar_in_item = note_start_bar - offset_bar
|
|
target_global_bar = item["start_bar"] + rel_bar_in_item
|
|
midi_events.append({
|
|
"note": note["pitch"],
|
|
"start_beat": target_global_bar * time_sig_num,
|
|
"duration_beats": note["duration_beats"],
|
|
"velocity": int(note.get("velocity", 0.8) * 127)
|
|
})
|
|
|
|
if midi_events:
|
|
try:
|
|
instrument_id = track.get("instrument", "")
|
|
plugin_mgr = PluginManager()
|
|
vst = plugin_mgr.load_vst(instrument_id) if instrument_id else None
|
|
|
|
instrument_source = track.get("instrument_source", "soundfont")
|
|
|
|
if instrument_source == "pianobook":
|
|
dspreset_path = track.get("dspreset_path", "")
|
|
if dspreset_path and os.path.exists(dspreset_path) and HAS_PEDALBOARD:
|
|
from pedalboard import Pedalboard
|
|
ds_manager = DecentSamplerManager()
|
|
try:
|
|
vst = ds_manager.create_decent_sampler_instance(dspreset_path)
|
|
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)
|
|
except Exception as e:
|
|
print(f"[RenderEngine] DecentSampler/Pianobook error: {e}")
|
|
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'
|
|
)
|
|
elif 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,
|
|
bank=soundfont_bank, program=soundfont_program
|
|
)
|
|
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(midi_channel, fid, soundfont_bank, soundfont_program)
|
|
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(midi_channel, 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(midi_channel, 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:
|
|
print(f"[RenderEngine] Error rendering MIDI: {e}")
|
|
|
|
elif item_type == "SECTION_ITEM":
|
|
source_data = item.get("source_data", {})
|
|
sec_id = source_data.get("referenced_section_id", "")
|
|
if sec_id and sec_id in section_store:
|
|
# Render nested section recursively
|
|
sec_container = section_store[sec_id]
|
|
sec_buffer = self.render_session_container(
|
|
session=sec_container,
|
|
section_store=section_store,
|
|
bpm=bpm,
|
|
time_sig_num=time_sig_num,
|
|
total_samples=total_samples
|
|
)
|
|
|
|
# Apply non-destructive crop/slicing on section buffer
|
|
if offset_sample < total_samples:
|
|
actual_dur = min(dur_samples, total_samples - offset_sample)
|
|
sliced_sec = sec_buffer[:, offset_sample : offset_sample + actual_dur]
|
|
|
|
# Write to track buffer
|
|
write_end = min(start_sample + sliced_sec.shape[1], total_samples)
|
|
actual_len = write_end - start_sample
|
|
if actual_len > 0:
|
|
track_buffer[:, start_sample:write_end] += sliced_sec[:, :actual_len]
|
|
|
|
# Apply Track Gain (via Pedalboard or fallback)
|
|
vol_db = track.get("volume_db", 0.0)
|
|
pan = track.get("pan", 0.0)
|
|
mute = track.get("mute", False)
|
|
|
|
if mute:
|
|
continue
|
|
|
|
# Apply Track FX (Chorus or Reverb)
|
|
fx_type = track.get("fx_type")
|
|
if fx_type == "chorus":
|
|
if HAS_PEDALBOARD:
|
|
try:
|
|
board = Pedalboard([Chorus(rate_hz=1.5, depth=0.25)])
|
|
track_buffer = board(track_buffer, sample_rate=self.sample_rate)
|
|
except Exception as e:
|
|
print(f"[RenderEngine] Pedalboard Chorus failed: {e}")
|
|
else:
|
|
# Fallback chorus using simple LFO delay modulation in scipy/numpy
|
|
try:
|
|
# 1.5 Hz sine LFO, modulating delay time between 15ms and 25ms (average 20ms)
|
|
lfo = 0.020 + 0.005 * np.sin(2 * np.pi * 1.5 * np.arange(total_samples) / self.sample_rate)
|
|
dry = track_buffer * 0.6
|
|
wet = np.zeros_like(track_buffer)
|
|
for ch in range(2):
|
|
indices = np.arange(total_samples) - (lfo * self.sample_rate)
|
|
indices = np.clip(indices, 0, total_samples - 1).astype(np.int32)
|
|
wet[ch, :] = track_buffer[ch, indices]
|
|
track_buffer = dry + wet * 0.5
|
|
except Exception as e:
|
|
print(f"[RenderEngine] Fallback Chorus failed: {e}")
|
|
elif fx_type == "reverb":
|
|
if HAS_PEDALBOARD:
|
|
try:
|
|
board = Pedalboard([Reverb(room_size=0.5, wet_level=0.4, dry_level=0.6)])
|
|
track_buffer = board(track_buffer, sample_rate=self.sample_rate)
|
|
except Exception as e:
|
|
print(f"[RenderEngine] Pedalboard Reverb failed: {e}")
|
|
else:
|
|
# Fallback reverb using exponentially decaying noise room impulse response
|
|
try:
|
|
# Generate impulse response (decaying noise)
|
|
len_ir = int(self.sample_rate * 2.0)
|
|
t_ir = np.arange(len_ir) / self.sample_rate
|
|
decay = np.exp(-t_ir / 0.5)
|
|
ir_l = (np.random.rand(len_ir) * 2 - 1) * decay
|
|
ir_r = (np.random.rand(len_ir) * 2 - 1) * decay
|
|
|
|
dry = track_buffer * 0.6
|
|
wet = np.zeros_like(track_buffer)
|
|
for ch in range(2):
|
|
ir = ir_l if ch == 0 else ir_r
|
|
# Convolve
|
|
conv = signal.convolve(track_buffer[ch, :], ir, mode='full')[:total_samples]
|
|
wet[ch, :] = conv
|
|
track_buffer = dry + wet * 0.4
|
|
except Exception as e:
|
|
print(f"[RenderEngine] Fallback Reverb failed: {e}")
|
|
|
|
# Process track volume
|
|
if HAS_PEDALBOARD:
|
|
try:
|
|
board = Pedalboard([Gain(gain_db=vol_db)])
|
|
processed_track = board(track_buffer, sample_rate=self.sample_rate)
|
|
except Exception:
|
|
gain_linear = 10 ** (vol_db / 20.0)
|
|
processed_track = track_buffer * gain_linear
|
|
else:
|
|
gain_linear = 10 ** (vol_db / 20.0)
|
|
processed_track = track_buffer * gain_linear
|
|
|
|
# Apply Track Pan
|
|
if pan != 0.0:
|
|
# Constant power panning
|
|
theta = ((np.clip(pan, -1.0, 1.0) + 1.0) / 2.0) * (np.pi / 2.0)
|
|
processed_track[0, :] *= np.cos(theta)
|
|
processed_track[1, :] *= np.sin(theta)
|
|
|
|
# Mix track to session
|
|
session_buffer += processed_track
|
|
|
|
return session_buffer
|
|
|
|
def render_project(self, project_json: dict, output_filepath: str):
|
|
bpm = project_json["metadata"]["bpm"]
|
|
time_sig_num = project_json["metadata"].get("time_signature_numerator", 4)
|
|
main_session = project_json["main_session"]
|
|
section_store = project_json.get("section_store", {})
|
|
|
|
# Compute total project samples
|
|
total_bars = main_session.get("length_bars", 16.0)
|
|
total_samples = self.bars_to_samples(total_bars, bpm, time_sig_num)
|
|
|
|
# Render main session
|
|
master_buffer = self.render_session_container(
|
|
session=main_session,
|
|
section_store=section_store,
|
|
bpm=bpm,
|
|
time_sig_num=time_sig_num,
|
|
total_samples=total_samples
|
|
)
|
|
|
|
# Normalization to prevent clipping
|
|
max_peak = np.max(np.abs(master_buffer))
|
|
if max_peak > 1.0:
|
|
master_buffer /= max_peak
|
|
|
|
# Write final output file
|
|
sf.write(output_filepath, master_buffer.T, self.sample_rate)
|
|
return output_filepath
|