3dtours
cc8b286f6c
1) Folder picker dung WINDOW EXPLORER (khong prompt nhap tay):
- src-tauri/src/lib.rs: IPC bridge — thread watcher ipc/pick_dir.request
-> run_on_main_thread -> dialog().file().blocking_pick_folder()
(tauri-plugin-dialog = IFileDialog/Explorer) -> ghi pick_dir.response;
ghi marker tauri_bridge_ready luc setup.
- app/api/v1/plugins.py: POST /pick-dir (def sync -> threadpool, khong
auth) — uu tien Tauri bridge, fallback PowerShell FolderBrowserDialog
(Win) / osascript (macOS) / zenity-kdialog (Linux).
- app.jsx pickPluginFolder: 1) pickPluginDir (native) -> 2) __TAURI__
invoke -> 3) in-app browser -> 4) prompt (cuoi cung).
2) Nut Synth liet ke VSTi da scan (truoc day rong):
- Root: list_available() chi quet settings.VST_DIR (mac dinh
/opt/daw_engine/vst3) trong khi Plugin Manager scan plugin_dirs user.
- plugins.py list_plugins: gop _scan_vst_in_dirs(plugin_dirs) (file
.vst3/.dll/.so + folder X.vst3 Windows).
- vst_engine.py: PluginManager.extra_vst_dirs + _scan_plugins quet them
(ca folder .vst3) + get_plugin_manager doc plugin_dirs.json -> load_vst
tim thay plugin user scan khi render.
3) VU meter leak: section-tab play -> sang MAIN SESSION -> track MAIN van
animate theo am section.
- Root: VU tick fallback _sub_ (sub-node section co analyser) chay cho
ca canvas MAIN; startSubTabPlayback ghi node o key track MAIN.
- Fix: fallback _sub_ chi ap dung cho canvas SECTION (isSessVu); 2 trigger
piano-roll them prefix _sess_ theo st.parent_tab_id.
Verify: 86 tests pass; engine frozen STARTUP 1.35s, 1 engine, 0 spawn '-c';
pick-dir IPC mock tra dung path; scan VST user dirs OK.
411 lines
17 KiB
Python
411 lines
17 KiB
Python
# SonicForge Studio VST / VSTi Engine Service
|
|
import os
|
|
import json
|
|
import numpy as np
|
|
import functools
|
|
from ctypes import c_char_p
|
|
|
|
def midi_note_to_freq(note_number: int) -> float:
|
|
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:
|
|
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)
|
|
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)
|
|
attack = min(int(0.01 * sr), actual_len // 4)
|
|
release = min(int(0.05 * sr), actual_len // 4)
|
|
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
|
|
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 _module_available(name: str) -> bool:
|
|
"""Kiem tra module co san khong — KHONG spawn subprocess.
|
|
|
|
Truoc day dung subprocess.run([sys.executable, '-c', 'import X']) —
|
|
khi app dong goi (PyInstaller frozen), sys.executable = daw_engine.exe
|
|
-> subprocess chay CA ENGINE (bootloader bo qua '-c', chay desktop_engine)
|
|
-> moi lan check lai sinh ra engine moi -> de quy spawn vo han
|
|
(Task Manager day daw_engine, port 8000-8005 leo thang, load rat cham).
|
|
find_spec() nhanh (micro-giay) va hoat dong ca source lan frozen.
|
|
"""
|
|
import importlib.util
|
|
try:
|
|
return importlib.util.find_spec(name) is not None
|
|
except (ImportError, AttributeError, ValueError):
|
|
return False
|
|
|
|
|
|
def check_pedalboard_safe():
|
|
return _module_available("pedalboard")
|
|
|
|
|
|
def check_pyfluidsynth_safe():
|
|
return _module_available("fluidsynth")
|
|
|
|
HAS_PEDALBOARD = check_pedalboard_safe()
|
|
HAS_PYFLUIDSYNTH = check_pyfluidsynth_safe()
|
|
|
|
def ensure_pyfluidsynth():
|
|
global HAS_PYFLUIDSYNTH
|
|
if not HAS_PYFLUIDSYNTH:
|
|
HAS_PYFLUIDSYNTH = check_pyfluidsynth_safe()
|
|
return HAS_PYFLUIDSYNTH
|
|
|
|
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
|
|
|
|
|
|
# ── Module-level caches ──
|
|
_FLUID_CACHE = {} # path → (fluidsynth.FluidSynth, refcount)
|
|
_PLUGIN_MANAGER_INSTANCE = None
|
|
_PLUGIN_MANAGER_ARGS = None
|
|
_SF_INSTRUMENTS_CACHE = {} # sf_id → list[presets]
|
|
|
|
def _load_user_plugin_dirs() -> list:
|
|
"""Đọc plugin_dirs.json (Plugin Manager user chọn) — cùng file với
|
|
plugins.py (STORAGE_DIR/plugin_dirs.json). Không import plugins.py để
|
|
tránh vòng import (plugins.py import vst_engine)."""
|
|
try:
|
|
from app.config import settings as _st
|
|
path = os.path.join(_st.STORAGE_DIR, "plugin_dirs.json")
|
|
if os.path.exists(path):
|
|
with open(path, "r", encoding="utf-8") as f:
|
|
data = json.load(f)
|
|
return [d for d in (data.get("plugin_dirs") or []) if d]
|
|
except Exception:
|
|
pass
|
|
return []
|
|
|
|
|
|
def get_plugin_manager(vst_dir=None, sf_dir=None, upload_sf_dir=None) -> "PluginManager":
|
|
"""Singleton: reuse PluginManager when args match, else create new.
|
|
Default dirs từ settings (env/.env/docker-compose hoặc user override).
|
|
VST scan gộp thêm plugin_dirs user (Plugin Manager) — nút Synth phải liệt
|
|
kê được VSTi đã scan và load_vst phải tìm thấy chúng khi render."""
|
|
from app.config import settings as _st
|
|
vst_dir = vst_dir or _st.VST_DIR
|
|
sf_dir = sf_dir or _st.SOUNDFONT_DIR
|
|
upload_sf_dir = upload_sf_dir or _st.STORAGE_DIR + "/soundfonts"
|
|
extra = _load_user_plugin_dirs()
|
|
global _PLUGIN_MANAGER_INSTANCE, _PLUGIN_MANAGER_ARGS
|
|
args = (vst_dir, sf_dir, upload_sf_dir, tuple(extra))
|
|
if _PLUGIN_MANAGER_INSTANCE is not None and _PLUGIN_MANAGER_ARGS == args:
|
|
return _PLUGIN_MANAGER_INSTANCE
|
|
_PLUGIN_MANAGER_ARGS = args
|
|
_PLUGIN_MANAGER_INSTANCE = PluginManager(vst_dir, sf_dir, upload_sf_dir, extra_vst_dirs=extra)
|
|
return _PLUGIN_MANAGER_INSTANCE
|
|
|
|
def load_soundfont_cached(path: str):
|
|
"""Return a cached low-level FluidSynth instance for path, incrementing refcount.
|
|
|
|
Uses the CFFI binding API (new_fluid_synth / fluid_synth_sfload) — the same
|
|
API render_engine relies on. The high-level `FluidSynth()`/`Synth()` classes
|
|
do not exist in this binding, so they are never used here.
|
|
"""
|
|
global _FLUID_CACHE
|
|
if not HAS_PYFLUIDSYNTH:
|
|
return None
|
|
if path in _FLUID_CACHE:
|
|
fl, ref = _FLUID_CACHE[path]
|
|
_FLUID_CACHE[path] = (fl, ref + 1)
|
|
return fl
|
|
try:
|
|
import fluidsynth as _fs
|
|
_settings = _fs.new_fluid_settings()
|
|
_fs.fluid_settings_setnum(_settings, b'synth.sample-rate', 44100.0)
|
|
fl = _fs.new_fluid_synth(_settings)
|
|
font_id = _fs.fluid_synth_sfload(fl, path.encode("utf-8"), 1)
|
|
if font_id < 0:
|
|
_fs.delete_fluid_synth(fl)
|
|
return None
|
|
_fs.fluid_synth_program_select(fl, 0, font_id, 0, 0)
|
|
_FLUID_CACHE[path] = (fl, 1)
|
|
return fl
|
|
except Exception:
|
|
return None
|
|
|
|
def release_soundfont(path: str):
|
|
"""Decrement refcount; delete FluidSynth when count reaches 0."""
|
|
global _FLUID_CACHE
|
|
if path not in _FLUID_CACHE:
|
|
return
|
|
fl, ref = _FLUID_CACHE[path]
|
|
if ref <= 1:
|
|
try:
|
|
import fluidsynth as _fs
|
|
_fs.delete_fluid_synth(fl)
|
|
except Exception:
|
|
pass
|
|
del _FLUID_CACHE[path]
|
|
else:
|
|
_FLUID_CACHE[path] = (fl, ref - 1)
|
|
|
|
class PluginManager:
|
|
def __init__(self, vst_dir=None, sf_dir=None, upload_sf_dir=None, extra_vst_dirs=None):
|
|
from app.config import settings as _st
|
|
self.vst_dir = vst_dir or _st.VST_DIR
|
|
self.sf_dir = sf_dir or _st.SOUNDFONT_DIR
|
|
self.upload_sf_dir = upload_sf_dir
|
|
# Thư mục VST thêm (plugin_dirs user scan trong Plugin Manager) —
|
|
# list_available()/load_vst phải thấy VSTi user đã scan (bug: nút
|
|
# Synth chỉ quét vst_dir env mặc định /opt/daw_engine/vst3).
|
|
self.extra_vst_dirs = [d for d in (extra_vst_dirs or []) if d]
|
|
self._sf_scan_cache = None # cache for _scan_soundfonts()
|
|
|
|
def _scan_plugins(self) -> dict:
|
|
plugins = {}
|
|
for scan_dir in [self.vst_dir] + self.extra_vst_dirs:
|
|
if not scan_dir or not os.path.isdir(scan_dir):
|
|
continue
|
|
for root, dirs, files in os.walk(scan_dir):
|
|
# Windows: VST3 là FOLDER tên X.vst3 (chứa X.vst3.dll bên trong)
|
|
for d in list(dirs):
|
|
if d.lower().endswith(".vst3"):
|
|
plugins[os.path.splitext(d)[0]] = os.path.join(root, d)
|
|
for file in files:
|
|
low = file.lower()
|
|
if low.endswith(".vst3") or low.endswith(".so") or low.endswith(".dll"):
|
|
plugin_path = os.path.join(root, file)
|
|
plugin_name = os.path.splitext(file)[0]
|
|
if plugin_name not in plugins:
|
|
plugins[plugin_name] = plugin_path
|
|
return plugins
|
|
|
|
def _scan_soundfonts(self) -> list:
|
|
sf_map = {}
|
|
dirs = [("system", self.sf_dir)]
|
|
if self.upload_sf_dir and self.upload_sf_dir != self.sf_dir:
|
|
dirs.append(("upload", self.upload_sf_dir))
|
|
|
|
meta_cache = {}
|
|
if self.upload_sf_dir and os.path.isdir(self.upload_sf_dir):
|
|
for f in os.listdir(self.upload_sf_dir):
|
|
if f.endswith(".meta"):
|
|
try:
|
|
import json
|
|
with open(os.path.join(self.upload_sf_dir, f), "r") as mf:
|
|
meta_cache[os.path.splitext(f)[0]] = json.load(mf)
|
|
except Exception:
|
|
pass
|
|
|
|
for source, d in dirs:
|
|
if not os.path.isdir(d):
|
|
continue
|
|
for f in os.listdir(d):
|
|
if f.endswith(".sf2") or f.endswith(".sf3"):
|
|
base_id = os.path.splitext(f)[0]
|
|
if base_id in sf_map:
|
|
continue
|
|
meta = meta_cache.get(base_id, None)
|
|
if meta:
|
|
display_name = meta.get("original_name", f)
|
|
else:
|
|
short_id = base_id[:8] if len(base_id) > 8 else base_id
|
|
display_name = f"SoundFont_{short_id}"
|
|
sf_map[base_id] = {
|
|
"id": base_id,
|
|
"name": display_name,
|
|
"file": f,
|
|
"display": os.path.splitext(display_name)[0][:40],
|
|
"source": source
|
|
}
|
|
return list(sf_map.values())
|
|
|
|
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 _scan_soundfonts_cached(self):
|
|
if self._sf_scan_cache is not None:
|
|
return self._sf_scan_cache
|
|
self._sf_scan_cache = self._scan_soundfonts()
|
|
return self._sf_scan_cache
|
|
|
|
def load_soundfont(self, path: str):
|
|
return load_soundfont_cached(path)
|
|
|
|
def list_soundfont_instruments(self, sf_id: str):
|
|
if not ensure_pyfluidsynth():
|
|
return []
|
|
if sf_id in _SF_INSTRUMENTS_CACHE:
|
|
return _SF_INSTRUMENTS_CACHE[sf_id]
|
|
search_dirs = []
|
|
if os.path.isdir(self.sf_dir):
|
|
search_dirs.append(self.sf_dir)
|
|
if self.upload_sf_dir and os.path.isdir(self.upload_sf_dir) and self.upload_sf_dir != self.sf_dir:
|
|
search_dirs.append(self.upload_sf_dir)
|
|
for d in search_dirs:
|
|
for f in os.listdir(d):
|
|
if not (f.endswith(".sf2") or f.endswith(".sf3")):
|
|
continue
|
|
base = os.path.splitext(f)[0]
|
|
if base == sf_id or base == sf_id.replace("sf_", ""):
|
|
path = os.path.join(d, f)
|
|
try:
|
|
import fluidsynth as _fs
|
|
# Low-level CFFI API (same as render_engine); never use
|
|
# the high-level Synth() class that this binding lacks.
|
|
_settings = _fs.new_fluid_settings()
|
|
_synth = _fs.new_fluid_synth(_settings)
|
|
try:
|
|
fid = _fs.fluid_synth_sfload(_synth, path.encode("utf-8"), 1)
|
|
if fid < 0:
|
|
continue
|
|
sfont = _fs.fluid_synth_get_sfont_by_id(_synth, fid)
|
|
presets = []
|
|
if sfont:
|
|
for bank in range(0, 2):
|
|
for prog_num in range(0, 128):
|
|
try:
|
|
preset = _fs.fluid_sfont_get_preset(sfont, bank, prog_num)
|
|
except Exception:
|
|
break
|
|
if preset:
|
|
try:
|
|
name_ptr = _fs.fluid_preset_get_name(preset)
|
|
if name_ptr:
|
|
if hasattr(_fs, "ffi"):
|
|
raw = _fs.ffi.string(name_ptr)
|
|
else:
|
|
raw = c_char_p(name_ptr).value
|
|
if raw:
|
|
presets.append({
|
|
"bank": bank,
|
|
"program": prog_num,
|
|
"name": raw.decode("utf-8", errors="replace")
|
|
})
|
|
except Exception:
|
|
continue
|
|
_SF_INSTRUMENTS_CACHE[sf_id] = presets[:256]
|
|
return presets[:256]
|
|
finally:
|
|
try:
|
|
_fs.delete_fluid_synth(_synth)
|
|
except Exception:
|
|
pass
|
|
except Exception:
|
|
import traceback; traceback.print_exc()
|
|
_SF_INSTRUMENTS_CACHE[sf_id] = []
|
|
return []
|
|
|
|
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, bank: int = None, program: int = None) -> list:
|
|
if not HAS_PEDALBOARD:
|
|
return []
|
|
beat_duration_sec = 60.0 / max(30.0, bpm)
|
|
messages = []
|
|
if bank is not None:
|
|
messages.append(MidiMessage(control_change=0, value=bank, sample_offset=0))
|
|
if program is not None:
|
|
messages.append(MidiMessage(program_change=program, sample_offset=0))
|
|
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
|
|
|
|
|
|
class DecentSamplerManager:
|
|
def __init__(self, vst_path="/opt/daw_engine/vst3/DecentSampler.vst3"):
|
|
self.vst_path = vst_path
|
|
|
|
def create_decent_sampler_instance(self, dspreset_path: str):
|
|
if not HAS_PEDALBOARD:
|
|
raise RuntimeError("pedalboard not available")
|
|
if not os.path.exists(self.vst_path):
|
|
raise FileNotFoundError(f"DecentSampler VST3 not found at {self.vst_path}")
|
|
if not os.path.exists(dspreset_path):
|
|
raise FileNotFoundError(f"Preset file not found at {dspreset_path}")
|
|
|
|
plugin = VST3Plugin(self.vst_path)
|
|
|
|
abs_preset = os.path.abspath(dspreset_path)
|
|
preset_dir = os.path.dirname(abs_preset)
|
|
cwd_before = os.getcwd()
|
|
try:
|
|
os.chdir(preset_dir)
|
|
plugin.load_preset(abs_preset)
|
|
finally:
|
|
os.chdir(cwd_before)
|
|
|
|
return plugin
|