3dtours
d40ac47d65
Scanner: new instance with existing sf_scan_state.json did not populate in-memory _catalog. Now re-inspects unchanged files to fill catalog on first scan. Frontend: left column reads instrumentSelectorData. soundfonts instead of sfPresets (null when no presets). Click SF in left column triggers on-demand fetch.
130 lines
4.5 KiB
Python
130 lines
4.5 KiB
Python
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 _inspect_single(self, fname: str, full: str, inspector) -> dict:
|
|
if fname.lower().endswith(".sf2"):
|
|
sf_info = inspector.inspect_sf2_file(full) or {}
|
|
else:
|
|
sf_info = {}
|
|
if not sf_info.get("soundfont_id"):
|
|
sf_id = os.path.splitext(fname)[0].lower()
|
|
sf_info = {
|
|
"soundfont_id": sf_id,
|
|
"filename": fname,
|
|
"total_instruments": 0,
|
|
"instruments": [],
|
|
"_sf3": True
|
|
}
|
|
return sf_info
|
|
|
|
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)
|
|
unchanged = prev and prev["size"] == sig[0] and prev["mtime"] == sig[1]
|
|
if unchanged and sf_id in self._catalog:
|
|
continue
|
|
if unchanged:
|
|
with self._lock:
|
|
if sf_id in self._catalog:
|
|
continue
|
|
sf_info = self._inspect_single(fname, full, inspector)
|
|
self._catalog[sf_id] = sf_info
|
|
continue
|
|
found_new = True
|
|
logger.info("New/changed SF detected: %s", fname)
|
|
sf_info = self._inspect_single(fname, full, inspector)
|
|
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 []
|