SonicForgeStudio/app/core/soundfont_inspector.py

import os
import json
import logging

logger = logging.getLogger(__name__)

try:
    from sf2utils.sf2parse import Sf2File
    HAS_SF2UTILS = True
except ImportError:
    HAS_SF2UTILS = False

GM_CATEGORIES = [
    ("Piano", range(0, 8)),
    ("Chromatic Percussion", range(8, 16)),
    ("Organ", range(16, 24)),
    ("Guitar", range(24, 32)),
    ("Bass", range(32, 40)),
    ("Strings", range(40, 48)),
    ("Ensemble", range(48, 56)),
    ("Brass", range(56, 64)),
    ("Reed", range(64, 72)),
    ("Pipe", range(72, 80)),
    ("Synth Lead", range(80, 90)),
    ("Synth Pad", range(90, 104)),
]

MAX_CONDENSED_ENTRIES = 50


class SoundFontInspector:
    def __init__(self, system_sf_dir="/opt/daw_engine/soundfonts", upload_sf_dir=None):
        self.system_sf_dir = system_sf_dir
        self.upload_sf_dir = upload_sf_dir
        self._catalog_cache = None

    def invalidate_catalog_cache(self):
        self._catalog_cache = None

    def inspect_sf2_file(self, filepath: str) -> dict:
        if not HAS_SF2UTILS:
            logger.warning("sf2utils not installed, cannot inspect .sf2 files")
            return {}
        if not os.path.exists(filepath):
            return {}

        try:
            sf_name = os.path.basename(filepath)
            sf_id = os.path.splitext(sf_name)[0].lower()

            instruments = []
            with open(filepath, 'rb') as f:
                sf2 = Sf2File(f)
                for preset in sf2.presets:
                    name = preset.name.strip()
                    if name == "EOP" or (preset.bank == 128 and preset.preset == 127):
                        continue
                    instruments.append({
                        "bank": preset.bank,
                        "program": preset.preset,
                        "name": name,
                        "is_percussion": (preset.bank == 128)
                    })

            return {
                "soundfont_id": sf_id,
                "filename": sf_name,
                "total_instruments": len(instruments),
                "instruments": instruments
            }
        except Exception as e:
            logger.warning(f"Skipping corrupted .sf2 file {filepath}: {e}")
            return {}

    def _scan_directory(self, directory: str) -> dict:
        catalog = {}
        if not os.path.isdir(directory):
            return catalog
        for fname in os.listdir(directory):
            if not fname.lower().endswith(('.sf2', '.sf3')):
                continue
            full_path = os.path.join(directory, fname)
            sf_info = self.inspect_sf2_file(full_path)
            if sf_info and sf_info.get("soundfont_id"):
                catalog[sf_info["soundfont_id"]] = sf_info
        return catalog

    def generate_full_catalog(self, output_json_path: str = None) -> dict:
        catalog = {}
        catalog.update(self._scan_directory(self.system_sf_dir))
        if self.upload_sf_dir and os.path.isdir(self.upload_sf_dir):
            catalog.update(self._scan_directory(self.upload_sf_dir))

        if output_json_path:
            os.makedirs(os.path.dirname(output_json_path), exist_ok=True)
            with open(output_json_path, 'w', encoding='utf-8') as f:
                json.dump(catalog, f, ensure_ascii=False, indent=2)

        self._catalog_cache = catalog
        return catalog

    def get_catalog(self) -> dict:
        if self._catalog_cache is not None:
            return self._catalog_cache
        return self.generate_full_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", [])
            if not instruments:
                continue

            selected = []
            used_programs = set()
            for cat_name, prog_range in GM_CATEGORIES:
                cat_members = [
                    inst for inst in instruments
                    if inst["program"] in prog_range and not inst["is_percussion"]
                ]
                if cat_members:
                    representative = cat_members[0]
                    key = (representative["program"], representative["bank"])
                    if key not in used_programs:
                        used_programs.add(key)
                        selected.append(representative)

            drum_kits = [inst for inst in instruments if inst["is_percussion"]]
            for dk in drum_kits[:3]:
                key = (dk["program"], dk["bank"])
                if key not in used_programs:
                    used_programs.add(key)
                    selected.append(dk)

            if len(selected) > MAX_CONDENSED_ENTRIES:
                selected = selected[:MAX_CONDENSED_ENTRIES]

            condensed[sf_id] = {
                "soundfont_id": sf_info["soundfont_id"],
                "filename": sf_info["filename"],
                "total_instruments": sf_info["total_instruments"],
                "condensed_count": len(selected),
                "instruments": selected
            }
        return condensed

    def format_condensed_for_prompt(self) -> str:
        condensed = self.get_condensed_catalog_summary()
        lines = []
        for sf_id, info in condensed.items():
            lines.append(f"SoundFont ID: '{sf_id}' (File: {info['filename']}):")
            for inst in info["instruments"]:
                lines.append(f"  - {inst['name']}: bank={inst['bank']}, program={inst['program']}")
        return "\n".join(lines)