3dtours/SonicForgeStudio
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: bổ sung MIDI

Browse Source
This commit is contained in:
3dtours
2026-07-23 08:32:23 +07:00
parent 6545e1746e
commit 0e44e44ecb
15 changed files with 3304 additions and 12680 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/core/render_engine.py
+229
View File
@@ -0,0 +1,229 @@
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
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
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)
for track in session.get("tracks", []):
track_type = track.get("type", "AUDIO")
track_buffer = np.zeros((2, total_samples), dtype=np.float32)
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:
# 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
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
# 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