SonicForgeStudio/app/core/soundfont_converter.py

import os
import struct
import subprocess
import logging
import tempfile
import wave

logger = logging.getLogger(__name__)

SF_TARGET_DIRS = [
    "/opt/daw_engine/soundfonts",
]


def _read_chunks(data, offset, max_size=0):
    chunks = []
    pos = offset
    end = len(data) if max_size == 0 else offset + max_size
    while pos + 8 <= end:
        ck_id = data[pos:pos + 4]
        ck_size = struct.unpack("<I", data[pos + 4:pos + 8])[0]
        chunks.append((ck_id, pos + 8, ck_size, pos))
        pos += 8 + ck_size
        if ck_size % 2 == 1:
            pos += 1
    return chunks


class SoundFontConverter:
    def __init__(self, target_dirs=None):
        self.target_dirs = target_dirs or SF_TARGET_DIRS

    def _check_ffmpeg_ogg(self):
        try:
            r = subprocess.run(["ffmpeg", "-encoders"], capture_output=True, text=True, timeout=5)
            return "libvorbis" in r.stdout
        except Exception:
            return False

    def _sf2_to_sf3_python(self, sf2_path: str, sf3_path: str) -> bool:
        has_ogg = self._check_ffmpeg_ogg()
        if not has_ogg:
            logger.warning("ffmpeg with libvorbis not available, cannot convert to SF3")
            return False

        try:
            with open(sf2_path, "rb") as f:
                data = f.read()
        except Exception as e:
            logger.error(f"Cannot read {sf2_path}: {e}")
            return False

        if len(data) < 12 or data[:4] != b"RIFF" or data[8:12] != b"sfbk":
            return False

        top_chunks = _read_chunks(data, 12)
        smpl_data = None
        smpl_offset = 0
        smpl_size = 0

        for ck_id, ck_data_off, ck_size, ck_head_off in top_chunks:
            if ck_id == b"LIST":
                list_type = data[ck_data_off:ck_data_off + 4]
                inner_chunks = _read_chunks(data, ck_data_off + 4, ck_size - 4)
                for ic_id, ic_data_off, ic_size, ic_head_off in inner_chunks:
                    if ic_id == b"smpl":
                        smpl_data = data[ic_data_off:ic_data_off + ic_size]
                        smpl_offset = ic_data_off
                        smpl_size = ic_size

        if smpl_data is None or smpl_size < 16:
            logger.warning("No sample data found in SF2")
            return False

        tmp_wav = sf3_path + ".tmp.wav"
        tmp_ogg = sf3_path + ".tmp.ogg"

        try:
            with open(tmp_wav, "wb") as fw:
                with wave.open(fw, "wb") as w:
                    w.setnchannels(1)
                    w.setsampwidth(2)
                    w.setframerate(44100)
                    w.writeframes(smpl_data)

            subprocess.run([
                "ffmpeg", "-y", "-i", tmp_wav,
                "-c:a", "libvorbis", "-q:a", "3",
                "-f", "ogg", tmp_ogg
            ], capture_output=True, timeout=600, check=True)

            with open(tmp_ogg, "rb") as fo:
                ogg_data = fo.read()

            compression = (1 - len(ogg_data) / max(len(smpl_data), 1)) * 100
            logger.info(f"Compressed {len(smpl_data)} -> {len(ogg_data)} bytes ({compression:.0f}%)")

            # Rebuild file: replace smpl chunk data with Ogg data
            ogg_padded = ogg_data if len(ogg_data) % 2 == 0 else ogg_data + b"\x00"
            new_smpl_size = len(ogg_data)

            out = bytearray()
            out.extend(data[:smpl_offset - 8])  # up to smpl chunk header
            out.extend(struct.pack("<I", new_smpl_size))  # new smpl size
            out.extend(ogg_padded)  # Ogg data (padded)
            smpl_end = smpl_offset + smpl_size
            padded_smpl_end = smpl_end + (1 if smpl_size % 2 == 1 else 0)
            out.extend(data[padded_smpl_end:])  # rest of file

            # Fix RIFF size
            old_total = struct.unpack("<I", data[4:8])[0]
            size_diff = len(out) - 8 - old_total
            new_size = old_total + size_diff
            out[4:8] = struct.pack("<I", new_size)

            with open(sf3_path, "wb") as fout:
                fout.write(bytes(out))

            return os.path.exists(sf3_path)
        except subprocess.TimeoutExpired:
            logger.error("Ogg conversion timed out")
            return False
        except Exception as e:
            logger.error(f"Conversion error: {e}")
            return False
        finally:
            for p in [tmp_wav, tmp_ogg]:
                try:
                    if os.path.exists(p): os.remove(p)
                except Exception:
                    pass

    @staticmethod
    def _find_sf3_converter():
        for exe in ["fluidsynth", "mscore"]:
            try:
                subprocess.run([exe, "--help"], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, timeout=2.0)
                return exe
            except (FileNotFoundError, subprocess.TimeoutExpired):
                continue
        return "python"

    def convert_sf2_to_sf3(self, sf2_path: str) -> str:
        if not os.path.exists(sf2_path):
            raise FileNotFoundError(f"Source SF2 file not found: {sf2_path}")

        sf3_path = os.path.splitext(sf2_path)[0] + ".sf3"

        if os.path.exists(sf3_path) and os.path.getmtime(sf3_path) >= os.path.getmtime(sf2_path):
            logger.info(f"SF3 already up-to-date: {sf3_path}")
            return sf3_path

        converter = self._find_sf3_converter()
        try:
            logger.info(f"Converting '{sf2_path}' -> '{sf3_path}' using {converter}...")
            if converter == "fluidsynth":
                logger.warning("fluidsynth CLI does not export SF3, using Python converter")
                converter = "python"

            if converter == "mscore":
                cmd = ["mscore", "-o", sf3_path, sf2_path]
                result = subprocess.run(cmd, capture_output=True, text=True, timeout=600)
                if result.returncode == 0 and os.path.exists(sf3_path):
                    logger.info(f"Created SF3 via mscore: {sf3_path} ({os.path.getsize(sf3_path)/1024/1024:.1f}MB)")
                    return sf3_path
                logger.warning(f"mscore failed, trying Python converter")
                converter = "python"

            if converter == "python":
                if self._sf2_to_sf3_python(sf2_path, sf3_path) and os.path.exists(sf3_path):
                    size_mb = os.path.getsize(sf3_path) / (1024 * 1024)
                    logger.info(f"Created SF3: {sf3_path} ({size_mb:.2f} MB)")
                    return sf3_path
                logger.warning(f"Python converter failed, returning SF2 path")
                return sf2_path

            return sf2_path
        except subprocess.TimeoutExpired:
            logger.error(f"Conversion timed out for {sf2_path}")
            return sf2_path
        except Exception as e:
            logger.error(f"Error converting {sf2_path}: {e}")
            return sf2_path

    def batch_convert_all(self):
        for sdir in self.target_dirs:
            if not os.path.isdir(sdir):
                continue
            for fname in sorted(os.listdir(sdir)):
                if fname.lower().endswith(".sf2"):
                    self.convert_sf2_to_sf3(os.path.join(sdir, fname))