ableton-mcp-ai/mcp_server/engines/libreria_analyzer.py

"""
LibreriaAnalyzer - Análisis espectral de samples de audio

Escanea recursivamente la librería de samples y extrae features espectrales
usando librosa (con fallback a scipy si no está disponible).

Uso:
    from engines.libreria_analyzer import LibreriaAnalyzer
    
    analyzer = LibreriaAnalyzer()
    analyzer.analyze_all()  # Analiza toda la librería
    
    # O consultar features de un sample específico
    features = analyzer.get_features("C:/.../kick_808.wav")
"""

import os
import json
import time
from pathlib import Path
from datetime import datetime, timedelta
from typing import Dict, List, Optional, Tuple, Any

# Audio analysis libraries
try:
    import numpy as np
    import librosa
    import librosa.feature
    LIBROSA_AVAILABLE = True
except ImportError:
    LIBROSA_AVAILABLE = False
    try:
        import numpy as np
        from scipy.io import wavfile
        from scipy import signal
        SCIPY_AVAILABLE = True
    except ImportError:
        SCIPY_AVAILABLE = False
        np = None


class LibreriaAnalyzer:
    """
    Analizador espectral de librería de samples.
    
    Extrae features de audio para todos los samples encontrados
    y los guarda en caché para evitar re-análisis.
    """
    
    # Extensiones de audio soportadas
    SUPPORTED_EXTENSIONS = {'.wav', '.mp3', '.aif', '.aiff', '.flac'}
    
    # Caché de features
    CACHE_FILENAME = '.features_cache.json'
    CACHE_MAX_AGE_DAYS = 7
    
    # Mapeo de carpetas a roles
    ROLE_MAPPING = {
        'kick': 'kick',
        'snare': 'snare',
        'bass': 'bass',
        'fx': 'fx',
        'drumloops': 'drum_loop',
        'drumloop': 'drum_loop',
        'hi-hat': 'hat_closed',
        'hihat': 'hat_closed',
        'hat': 'hat_closed',
        'oneshots': 'oneshot',
        'oneshot': 'oneshot',
        'perc loop': 'perc_loop',
        'perc_loop': 'perc_loop',
        'reggaeton 3': 'synth',
        'sentimientolatino2025': 'multi',
        'sounds presets': 'preset',
        'extra': 'extra',
        'flp': 'project',
    }
    
    def __init__(self, library_path: str = None, verbose: bool = True):
        """
        Inicializa el analizador.
        
        Args:
            library_path: Ruta base de la librería. Por defecto: libreria/reggaeton/
            verbose: Si True, muestra progreso del análisis
        """
        if library_path is None:
            # Default path según la estructura del proyecto
            base_path = Path("C:/ProgramData/Ableton/Live 12 Suite/Resources/MIDI Remote Scripts")
            self.library_path = base_path / "libreria" / "reggaeton"
        else:
            self.library_path = Path(library_path)
        
        self.verbose = verbose
        self.features: Dict[str, Dict[str, Any]] = {}
        self.cache_path = self.library_path / self.CACHE_FILENAME
        
        # Verificar disponibilidad de librerías
        if not LIBROSA_AVAILABLE and not SCIPY_AVAILABLE:
            raise ImportError(
                "Se requiere librosa o scipy para análisis de audio. "
                "Instala: pip install librosa numpy"
            )
        
        # Cargar caché existente si está disponible
        self._load_cache()
    
    def _load_cache(self) -> bool:
        """
        Carga el caché de features si existe y es reciente.
        
        Returns:
            True si se cargó el caché, False en caso contrario
        """
        if not self.cache_path.exists():
            return False
        
        try:
            # Verificar edad del caché
            cache_age = datetime.now() - datetime.fromtimestamp(
                self.cache_path.stat().st_mtime
            )
            
            if cache_age > timedelta(days=self.CACHE_MAX_AGE_DAYS):
                if self.verbose:
                    print(f"[LibreriaAnalyzer] Caché expirado ({cache_age.days} días). Re-analizando...")
                return False
            
            # Cargar caché
            with open(self.cache_path, 'r', encoding='utf-8') as f:
                cache_data = json.load(f)
            
            self.features = cache_data.get('samples', {})
            
            if self.verbose:
                total = cache_data.get('total_samples', len(self.features))
                scan_date = cache_data.get('scan_date', 'unknown')
                print(f"[LibreriaAnalyzer] Caché cargado: {total} samples (desde {scan_date})")
            
            return True
            
        except (json.JSONDecodeError, IOError, KeyError) as e:
            if self.verbose:
                print(f"[LibreriaAnalyzer] Error cargando caché: {e}")
            return False
    
    def _save_cache(self) -> None:
        """Guarda las features actuales en el caché."""
        cache_data = {
            "version": "1.0",
            "total_samples": len(self.features),
            "scan_date": datetime.now().isoformat(),
            "library_path": str(self.library_path),
            "samples": self.features
        }
        
        try:
            with open(self.cache_path, 'w', encoding='utf-8') as f:
                json.dump(cache_data, f, indent=2, ensure_ascii=False)
            
            if self.verbose:
                print(f"[LibreriaAnalyzer] Caché guardado: {len(self.features)} samples")
        except IOError as e:
            if self.verbose:
                print(f"[LibreriaAnalyzer] Error guardando caché: {e}")
    
    def _detect_role(self, file_path: Path) -> str:
        """
        Detecta el rol del sample basado en la carpeta contenedora.
        
        Args:
            file_path: Ruta al archivo de audio
            
        Returns:
            Rol detectado (kick, snare, bass, etc.)
        """
        # Obtener partes del path en minúsculas
        path_parts = [p.lower() for p in file_path.parts]
        
        # Buscar coincidencias en el mapeo
        for part in path_parts:
            # Remover caracteres especiales para matching
            clean_part = part.replace(' ', '_').replace('-', '_').replace('(', '').replace(')', '')
            
            if part in self.ROLE_MAPPING:
                return self.ROLE_MAPPING[part]
            if clean_part in self.ROLE_MAPPING:
                return self.ROLE_MAPPING[clean_part]
            
            # Buscar substrings
            for key, role in self.ROLE_MAPPING.items():
                if key in part or key in clean_part:
                    return role
        
        return "unknown"
    
    def _get_pack_name(self, file_path: Path) -> str:
        """
        Obtiene el nombre del pack/carpeta padre del sample.
        
        Args:
            file_path: Ruta al archivo de audio
            
        Returns:
            Nombre del pack/carpeta
        """
        # El pack es el directorio padre inmediato
        parent = file_path.parent.name
        return parent if parent else "root"
    
    def _extract_features_librosa(self, file_path: Path) -> Optional[Dict[str, Any]]:
        """
        Extrae features de audio usando librosa.
        
        Args:
            file_path: Ruta al archivo de audio
            
        Returns:
            Diccionario con features o None si hay error
        """
        try:
            # Cargar audio
            y, sr = librosa.load(str(file_path), sr=None, mono=True)
            
            # Duración
            duration = librosa.get_duration(y=y, sr=sr)
            
            # RMS (energía)
            rms = float(np.mean(librosa.feature.rms(y=y)))
            rms_db = 20 * np.log10(rms + 1e-10)  # Convertir a dB
            
            # Spectral Centroid (brillo)
            spectral_centroid = float(np.mean(librosa.feature.spectral_centroid(y=y, sr=sr)))
            
            # Spectral Rolloff
            spectral_rolloff = float(np.mean(librosa.feature.spectral_rolloff(y=y, sr=sr)))
            
            # Zero Crossing Rate
            zcr = float(np.mean(librosa.feature.zero_crossing_rate(y)))
            
            # MFCCs (13 coeficientes)
            mfccs = librosa.feature.mfcc(y=y, sr=sr, n_mfcc=13)
            mfccs_mean = [float(np.mean(coef)) for coef in mfccs]
            
            # Onset Strength (qué tan rítmico es)
            onset_env = librosa.onset.onset_strength(y=y, sr=sr)
            onset_strength = float(np.mean(onset_env))
            
            # BPM detection
            try:
                tempo, _ = librosa.beat.beat_track(y=y, sr=sr)
                bpm = float(tempo) if isinstance(tempo, (int, float, np.number)) else float(tempo[0])
            except:
                bpm = 0.0
            
            # Key detection via chromagram
            try:
                chromagram = librosa.feature.chroma_cqt(y=y, sr=sr)
                # Sumar a lo largo del tiempo para obtener el perfil de pitch
                chroma_avg = np.sum(chromagram, axis=1)
                # Notas musicales
                notes = ['C', 'C#', 'D', 'D#', 'E', 'F', 'F#', 'G', 'G#', 'A', 'A#', 'B']
                # Encontrar la nota dominante
                key_index = np.argmax(chroma_avg)
                key = notes[key_index]
                
                # Detectar si es mayor o menor (heurística simple)
                # Si el tercer grado está presente, es menor
                minor_third_idx = (key_index + 3) % 12
                if chroma_avg[minor_third_idx] > chroma_avg[(key_index + 4) % 12]:
                    key += 'm'
            except:
                key = ""
            
            # Determinar canales (asumimos mono después de librosa.load con mono=True)
            # Para saber si era stereo originalmente, tendríamos que cargar de nuevo
            try:
                y_orig, _ = librosa.load(str(file_path), sr=None, mono=False)
                channels = y_orig.shape[0] if len(y_orig.shape) > 1 else 1
            except:
                channels = 1
            
            return {
                "rms": round(rms_db, 2),
                "spectral_centroid": round(spectral_centroid, 2),
                "spectral_rolloff": round(spectral_rolloff, 2),
                "zero_crossing_rate": round(zcr, 4),
                "mfccs": [round(m, 4) for m in mfccs_mean],
                "onset_strength": round(onset_strength, 4),
                "duration": round(duration, 3),
                "sample_rate": sr,
                "channels": channels,
                "bpm": round(bpm, 1) if bpm > 0 else 0,
                "key": key
            }
            
        except Exception as e:
            if self.verbose:
                print(f"[LibreriaAnalyzer] Error analizando {file_path}: {e}")
            return None
    
    def _extract_features_scipy(self, file_path: Path) -> Optional[Dict[str, Any]]:
        """
        Extrae features básicas usando scipy (fallback cuando librosa no está).
        
        Solo soporta archivos WAV.
        
        Args:
            file_path: Ruta al archivo de audio
            
        Returns:
            Diccionario con features básicas o None si hay error
        """
        try:
            # scipy solo soporta WAV nativamente
            if file_path.suffix.lower() not in {'.wav'}:
                return None
            
            # Cargar audio
            sr, data = wavfile.read(str(file_path))
            
            # Convertir a float y mono si es necesario
            if data.ndim > 1:
                channels = data.shape[1]
                data = np.mean(data, axis=1)  # Convertir a mono
            else:
                channels = 1
            
            # Normalizar a float [-1, 1]
            if data.dtype == np.int16:
                data = data.astype(np.float32) / 32768.0
            elif data.dtype == np.int32:
                data = data.astype(np.float32) / 2147483648.0
            else:
                data = data.astype(np.float32)
            
            # Duración
            duration = len(data) / sr
            
            # RMS
            rms = np.sqrt(np.mean(data ** 2))
            rms_db = 20 * np.log10(rms + 1e-10)
            
            # Spectral Centroid usando FFT
            fft = np.fft.fft(data)
            freqs = np.fft.fftfreq(len(data), 1/sr)
            magnitude = np.abs(fft)
            
            # Solo frecuencias positivas
            positive_freqs = freqs[:len(freqs)//2]
            positive_magnitude = magnitude[:len(magnitude)//2]
            
            spectral_centroid = np.sum(positive_freqs * positive_magnitude) / np.sum(positive_magnitude)
            
            # Zero Crossing Rate
            zcr = np.mean(np.diff(np.sign(data)) != 0)
            
            # No podemos hacer análisis avanzado sin librosa
            return {
                "rms": round(rms_db, 2),
                "spectral_centroid": round(float(spectral_centroid), 2),
                "spectral_rolloff": 0.0,  # No disponible sin librosa
                "zero_crossing_rate": round(float(zcr), 4),
                "mfccs": [],  # No disponible sin librosa
                "onset_strength": 0.0,  # No disponible sin librosa
                "duration": round(duration, 3),
                "sample_rate": sr,
                "channels": channels,
                "bpm": 0,  # No disponible sin librosa
                "key": ""  # No disponible sin librosa
            }
            
        except Exception as e:
            if self.verbose:
                print(f"[LibreriaAnalyzer] Error (scipy) analizando {file_path}: {e}")
            return None
    
    def _extract_features(self, file_path: Path) -> Optional[Dict[str, Any]]:
        """
        Extrae features de un archivo de audio.
        
        Usa librosa si está disponible, de lo contrario usa scipy.
        
        Args:
            file_path: Ruta al archivo de audio
            
        Returns:
            Diccionario con features o None si hay error
        """
        if LIBROSA_AVAILABLE:
            return self._extract_features_librosa(file_path)
        elif SCIPY_AVAILABLE:
            return self._extract_features_scipy(file_path)
        else:
            return None
    
    def _scan_samples(self) -> List[Path]:
        """
        Escanea recursivamente la librería buscando samples de audio.
        
        Returns:
            Lista de rutas a archivos de audio encontrados
        """
        samples = []
        
        if not self.library_path.exists():
            if self.verbose:
                print(f"[LibreriaAnalyzer] Librería no encontrada: {self.library_path}")
            return samples
        
        for ext in self.SUPPORTED_EXTENSIONS:
            samples.extend(self.library_path.rglob(f"*{ext}"))
        
        return samples
    
    def analyze_sample(self, file_path: str) -> Optional[Dict[str, Any]]:
        """
        Analiza un sample individual y extrae sus features.
        
        Args:
            file_path: Ruta al archivo de audio
            
        Returns:
            Diccionario con todas las features del sample
        """
        path = Path(file_path)
        
        if not path.exists():
            if self.verbose:
                print(f"[LibreriaAnalyzer] Archivo no encontrado: {file_path}")
            return None
        
        if path.suffix.lower() not in self.SUPPORTED_EXTENSIONS:
            if self.verbose:
                print(f"[LibreriaAnalyzer] Formato no soportado: {path.suffix}")
            return None
        
        # Extraer features de audio
        audio_features = self._extract_features(path)
        
        if audio_features is None:
            return None
        
        # Construir el objeto completo de features
        abs_path = str(path.resolve())
        role = self._detect_role(path)
        pack = self._get_pack_name(path)
        
        features = {
            "name": path.name,
            "pack": pack,
            "role": role,
            **audio_features
        }
        
        # Guardar en caché interno
        self.features[abs_path] = features
        
        return features
    
    def analyze_all(self, force_reanalyze: bool = False) -> Dict[str, Dict[str, Any]]:
        """
        Analiza todos los samples de la librería.
        
        Args:
            force_reanalyze: Si True, re-analiza incluso si hay caché
            
        Returns:
            Diccionario con todas las features indexadas por path
        """
        # Verificar si ya tenemos caché válido
        if not force_reanalyze and self.features:
            if self.verbose:
                print(f"[LibreriaAnalyzer] Usando caché existente con {len(self.features)} samples")
            return self.features
        
        # Escanear samples
        samples = self._scan_samples()
        
        if not samples:
            if self.verbose:
                print(f"[LibreriaAnalyzer] No se encontraron samples en {self.library_path}")
            return {}
        
        if self.verbose:
            print(f"[LibreriaAnalyzer] Encontrados {len(samples)} samples para analizar")
        
        # Analizar cada sample
        total = len(samples)
        analyzed = 0
        failed = 0
        
        for i, sample_path in enumerate(samples, 1):
            abs_path = str(sample_path.resolve())
            
            # Verificar si ya está en caché y no es force_reanalyze
            if not force_reanalyze and abs_path in self.features:
                continue
            
            # Analizar sample
            features = self.analyze_sample(abs_path)
            
            if features:
                analyzed += 1
            else:
                failed += 1
            
            # Mostrar progreso
            if self.verbose and i % 10 == 0:
                pct = (i / total) * 100
                print(f"[LibreriaAnalyzer] Progreso: {i}/{total} ({pct:.1f}%) - OK: {analyzed}, Fallos: {failed}")
        
        if self.verbose:
            print(f"[LibreriaAnalyzer] Análisis completo: {analyzed} analizados, {failed} fallidos")
        
        # Guardar caché
        self._save_cache()
        
        return self.features
    
    def get_features(self, sample_path: str) -> Optional[Dict[str, Any]]:
        """
        Obtiene las features de un sample específico.
        
        Si el sample no está en caché, lo analiza.
        
        Args:
            sample_path: Ruta al archivo de audio
            
        Returns:
            Diccionario con features o None si no se puede analizar
        """
        abs_path = str(Path(sample_path).resolve())
        
        # Verificar si está en caché
        if abs_path in self.features:
            return self.features[abs_path]
        
        # Analizar si no está en caché
        return self.analyze_sample(sample_path)
    
    def get_all_features(self) -> Dict[str, Dict[str, Any]]:
        """
        Obtiene todas las features cargadas/analizadas.
        
        Returns:
            Diccionario con todas las features
        """
        return self.features
    
    def clear_cache(self) -> None:
        """Elimina el archivo de caché y limpia las features en memoria."""
        self.features = {}
        if self.cache_path.exists():
            try:
                self.cache_path.unlink()
                if self.verbose:
                    print(f"[LibreriaAnalyzer] Caché eliminado: {self.cache_path}")
            except IOError as e:
                if self.verbose:
                    print(f"[LibreriaAnalyzer] Error eliminando caché: {e}")
    
    def get_stats(self) -> Dict[str, Any]:
        """
        Obtiene estadísticas de la librería analizada.
        
        Returns:
            Diccionario con estadísticas
        """
        if not self.features:
            return {
                "total_samples": 0,
                "by_role": {},
                "avg_duration": 0,
                "avg_rms": 0
            }
        
        # Contar por rol
        by_role = {}
        total_duration = 0
        total_rms = 0
        
        for path, features in self.features.items():
            role = features.get("role", "unknown")
            by_role[role] = by_role.get(role, 0) + 1
            
            total_duration += features.get("duration", 0)
            total_rms += features.get("rms", 0)
        
        total = len(self.features)
        
        return {
            "total_samples": total,
            "by_role": by_role,
            "avg_duration": round(total_duration / total, 3) if total > 0 else 0,
            "avg_rms": round(total_rms / total, 2) if total > 0 else 0
        }


# Función de conveniencia para uso directo
def analyze_library(library_path: str = None, verbose: bool = True) -> LibreriaAnalyzer:
    """
    Analiza toda la librería y retorna el analizador configurado.
    
    Args:
        library_path: Ruta a la librería (default: libreria/reggaeton/)
        verbose: Mostrar progreso
        
    Returns:
        Instancia de LibreriaAnalyzer con todas las features cargadas
    """
    analyzer = LibreriaAnalyzer(library_path=library_path, verbose=verbose)
    analyzer.analyze_all()
    return analyzer


if __name__ == "__main__":
    # Test básico
    print("[LibreriaAnalyzer] Test de inicialización...")
    
    try:
        analyzer = LibreriaAnalyzer(verbose=True)
        print(f"Librería: {analyzer.library_path}")
        print(f"Caché: {analyzer.cache_path}")
        print(f"Librosa disponible: {LIBROSA_AVAILABLE}")
        print(f"Scipy disponible: {SCIPY_AVAILABLE}")
        
        # Intentar cargar/analizar
        features = analyzer.analyze_all()
        print(f"\nTotal samples en caché: {len(features)}")
        
        # Mostrar estadísticas
        stats = analyzer.get_stats()
        print(f"\nEstadísticas: {json.dumps(stats, indent=2)}")
        
    except Exception as e:
        print(f"Error: {e}")
        import traceback
        traceback.print_exc()