SonicForgeStudio/app/core/soundfont_converter.py

import os
import struct
import subprocess
import logging
import wave

logger = logging.getLogger(__name__)

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


def _find_chunk(data, chunk_id, offset=12, max_size=0):
    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]
        if ck_id == chunk_id:
            return (pos, ck_id, ck_size, pos + 8)
        if ck_id == b"LIST" and pos + 12 <= end:
            list_type = data[pos + 8:pos + 12]
            # Check if we're looking for a specific list type
            if chunk_id == b"smpl" and list_type == b"sdta":
                inner = _find_chunk_in_list(data, chunk_id, pos + 12, ck_size - 4)
                if inner:
                    return inner
            elif chunk_id == b"LIST":
                # When searching for LIST by type, check if it's sdta
                pass
            else:
                inner = _find_chunk_in_list(data, chunk_id, pos + 12, ck_size - 4)
                if inner:
                    return inner
        pos += 8 + ck_size
        if ck_size % 2 == 1:
            pos += 1
    return None


def _find_list_of_type(data, list_type_id, offset=12):
    pos = offset
    while pos + 12 <= len(data):
        ck_id = data[pos:pos + 4]
        ck_size = struct.unpack("<I", data[pos + 4:pos + 8])[0]
        if ck_id == b"LIST":
            form_type = data[pos + 8:pos + 12]
            if form_type == list_type_id:
                return (pos, ck_id, ck_size, pos + 12)
        pos += 8 + ck_size
        if ck_size % 2 == 1:
            pos += 1
    return None


def _find_chunk_in_list(data, chunk_id, list_data_offset, list_data_size):
    pos = list_data_offset
    end = list_data_offset + list_data_size
    while pos + 8 <= end:
        ck_id = data[pos:pos + 4]
        ck_size = struct.unpack("<I", data[pos + 4:pos + 8])[0]
        if ck_id == chunk_id:
            return (pos, ck_id, ck_size, pos + 8)
        pos += 8 + ck_size
        if ck_size % 2 == 1:
            pos += 1
    return None


def _update_size(data, offset, new_size):
    return data[:offset] + struct.pack("<I", new_size) + data[offset + 4:]


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":
            logger.warning("Not a valid SF2 file")
            return False

        # Find smpl chunk recursively
        smpl = _find_chunk(data, b"smpl")
        if smpl is None:
            logger.warning("No smpl chunk found in SF2")
            return False

        smpl_head_off, _, smpl_old_size, smpl_data_off = smpl
        sample_data = data[smpl_data_off:smpl_data_off + smpl_old_size]

        if len(sample_data) < 16:
            logger.warning("Sample data too small")
            return False

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

        try:
            # Write samples as WAV
            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(sample_data)

            # Compress to Ogg Vorbis
            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(sample_data), 1)) * 100
            logger.info(f"Compressed {len(sample_data)} -> {len(ogg_data)} bytes ({compression:.0f}%)")

            ogg_padded = ogg_data if len(ogg_data) % 2 == 0 else ogg_data + b"\x00"
            new_smpl_size = len(ogg_data)
            old_padded = smpl_old_size + (1 if smpl_old_size % 2 == 1 else 0)
            new_padded = len(ogg_padded)
            delta = new_padded - old_padded

            # Rebuild file: replace smpl chunk and update all sizes
            out = bytearray()
            out.extend(data[:smpl_head_off])           # up to smpl chunk header
            out.extend(struct.pack("<I", new_smpl_size))  # new smpl size
            out.extend(ogg_padded)                     # compressed data (even-padded)
            out.extend(data[smpl_head_off + 8 + old_padded:])  # rest of file

            data_out = bytes(out)

            # Find sdta LIST and update its size
            sdta = _find_list_of_type(data_out, b"sdta", 12)
            if sdta:
                lh_off, _, lh_size, ld_off = sdta
                data_out = _update_size(data_out, lh_off + 4, lh_size + delta)

            # Update RIFF root size
            old_riff_size = struct.unpack("<I", data_out[4:8])[0]
            new_total = len(data_out) - 8
            data_out = _update_size(data_out, 4, new_total)

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

            # Validate: check that RIFF size matches actual size
            written = os.path.getsize(sf3_path)
            parsed_riff = struct.unpack("<I", data_out[4:8])[0]
            if parsed_riff != written - 8:
                logger.warning(f"Size mismatch: RIFF says {parsed_riff}, actual is {written - 8}")
                return False

            return True

        except subprocess.TimeoutExpired:
            logger.error("Ogg conversion timed out")
            return False
        except Exception as e:
            logger.error(f"Conversion error: {e}")
            import traceback
            traceback.print_exc()
            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":
                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
                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 for {sf2_path}, 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))