feat: auto-scan soundfont for Synth instrument list
Background daemon (30s interval) tracks scanned files via sf_scan_state.json (size+mtime). Only inspects new/changed SFs. Catalog served from scanner cache, no full re-scan per request. Upload/delete trigger immediate re-scan.
This commit is contained in:
@@ -104,8 +104,8 @@ class SoundFontInspector:
|
||||
return self._catalog_cache
|
||||
return self.generate_full_catalog()
|
||||
|
||||
def get_condensed_catalog_summary(self) -> dict:
|
||||
catalog = self.get_catalog()
|
||||
def get_condensed_catalog_summary(self, catalog: dict = None) -> dict:
|
||||
catalog = catalog if catalog is not None else self.get_catalog()
|
||||
condensed = {}
|
||||
for sf_id, sf_info in catalog.items():
|
||||
instruments = sf_info.get("instruments", [])
|
||||
|
||||
@@ -0,0 +1,117 @@
|
||||
import os
|
||||
import json
|
||||
import time
|
||||
import logging
|
||||
import threading
|
||||
from app.config import settings
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
TRACK_FILE = os.path.join(settings.STORAGE_DIR, "sf_scan_state.json")
|
||||
SYSTEM_SF_DIR = "/opt/daw_engine/soundfonts"
|
||||
UPLOAD_SF_DIR = os.path.join(settings.STORAGE_DIR, "soundfonts")
|
||||
|
||||
|
||||
def _file_sig(path: str) -> tuple:
|
||||
s = os.path.getsize(path)
|
||||
m = os.path.getmtime(path)
|
||||
return (s, m)
|
||||
|
||||
|
||||
class SoundFontAutoScanner:
|
||||
def __init__(self, system_sf_dir=SYSTEM_SF_DIR, upload_sf_dir=UPLOAD_SF_DIR):
|
||||
self.system_sf_dir = system_sf_dir
|
||||
self.upload_sf_dir = upload_sf_dir
|
||||
self._catalog = {}
|
||||
self._lock = threading.Lock()
|
||||
self._state = self._load_state()
|
||||
|
||||
def _load_state(self) -> dict:
|
||||
if not os.path.exists(TRACK_FILE):
|
||||
return {}
|
||||
try:
|
||||
with open(TRACK_FILE) as f:
|
||||
return json.load(f)
|
||||
except Exception as e:
|
||||
logger.warning("scan_state load failed: %s", e)
|
||||
return {}
|
||||
|
||||
def _save_state(self):
|
||||
os.makedirs(os.path.dirname(TRACK_FILE), exist_ok=True)
|
||||
with open(TRACK_FILE, "w") as f:
|
||||
json.dump(self._state, f, indent=2)
|
||||
|
||||
def _sf_files(self, directory: str) -> list:
|
||||
if not os.path.isdir(directory):
|
||||
return []
|
||||
out = []
|
||||
for fname in os.listdir(directory):
|
||||
if fname.lower().endswith((".sf2", ".sf3")):
|
||||
out.append((fname, os.path.join(directory, fname)))
|
||||
return out
|
||||
|
||||
def scan_once(self) -> bool:
|
||||
from app.core.soundfont_inspector import SoundFontInspector
|
||||
inspector = SoundFontInspector(self.system_sf_dir, self.upload_sf_dir)
|
||||
found_new = False
|
||||
dirs = [(self.system_sf_dir, "system")]
|
||||
if self.upload_sf_dir and os.path.isdir(self.upload_sf_dir):
|
||||
dirs.append((self.upload_sf_dir, "upload"))
|
||||
|
||||
for directory, source in dirs:
|
||||
for fname, full in self._sf_files(directory):
|
||||
sf_id = os.path.splitext(fname)[0].lower()
|
||||
key = f"{source}:{sf_id}"
|
||||
sig = _file_sig(full)
|
||||
prev = self._state.get(key)
|
||||
if prev and prev["size"] == sig[0] and prev["mtime"] == sig[1]:
|
||||
if sf_id in self._catalog:
|
||||
continue
|
||||
found_new = True
|
||||
logger.info("New/changed SF detected: %s", fname)
|
||||
if fname.lower().endswith(".sf2"):
|
||||
sf_info = inspector.inspect_sf2_file(full) or {}
|
||||
else:
|
||||
sf_info = {}
|
||||
if not sf_info.get("soundfont_id"):
|
||||
sf_info = {
|
||||
"soundfont_id": sf_id,
|
||||
"filename": fname,
|
||||
"total_instruments": 0,
|
||||
"instruments": [],
|
||||
"_sf3": True
|
||||
}
|
||||
with self._lock:
|
||||
self._catalog[sf_id] = sf_info
|
||||
self._state[key] = {"size": sig[0], "mtime": sig[1], "file": fname}
|
||||
|
||||
if found_new:
|
||||
self._save_state()
|
||||
inspector.invalidate_catalog_cache()
|
||||
return found_new
|
||||
|
||||
def scan_loop(self, interval: int = 30, stop_event: threading.Event = None):
|
||||
logger.info("SF auto-scanner started (interval=%ds)", interval)
|
||||
self.scan_once()
|
||||
while not (stop_event and stop_event.is_set()):
|
||||
time.sleep(interval)
|
||||
try:
|
||||
self.scan_once()
|
||||
except Exception as e:
|
||||
logger.error("scan cycle error: %s", e)
|
||||
|
||||
def start_background(self, interval: int = 30) -> threading.Event:
|
||||
ev = threading.Event()
|
||||
t = threading.Thread(target=self.scan_loop, args=(interval, ev), daemon=True)
|
||||
t.start()
|
||||
return ev
|
||||
|
||||
def get_catalog(self) -> dict:
|
||||
with self._lock:
|
||||
return dict(self._catalog)
|
||||
|
||||
def get_instruments(self, sf_id: str) -> list:
|
||||
entry = self._catalog.get(sf_id)
|
||||
if entry:
|
||||
return entry.get("instruments", [])
|
||||
return []
|
||||
Reference in New Issue
Block a user