5ce8187c65
- Add _cmd_create_arrangement_audio_pattern with 5-method fallback chain - Method 1: track.insert_arrangement_clip() [Live 12+] - Method 2: track.create_audio_clip() [Live 11+] - Method 3: arrangement_clips.add_new_clip() [Live 12+] - Method 4: Session->duplicate_clip_to_arrangement [Legacy] - Method 5: Session->Recording [Universal] - Add _cmd_duplicate_clip_to_arrangement for session-to-arrangement workflow - Update skills documentation - Verified: 3 clips created at positions [0, 4, 8] in Arrangement View Closes: Audio injection in Arrangement View
923 lines
33 KiB
Python
923 lines
33 KiB
Python
"""
|
|
Reference Matcher - Analyzes reference tracks and creates user sound profiles.
|
|
|
|
Este módulo analiza archivos de referencia (como reggaeton_ejemplo.mp3),
|
|
extrae sus características espectrales y genera un perfil de sonido
|
|
personalizado para el usuario basado en samples similares de la librería.
|
|
"""
|
|
import json
|
|
import logging
|
|
import os
|
|
from pathlib import Path
|
|
from typing import Dict, List, Optional, Any, Tuple
|
|
from dataclasses import dataclass, field, asdict
|
|
import numpy as np
|
|
from collections import Counter
|
|
|
|
logger = logging.getLogger("ReferenceMatcher")
|
|
|
|
# Paths
|
|
LIBRERIA_DIR = Path(r"C:\ProgramData\Ableton\Live 12 Suite\Resources\MIDI Remote Scripts\libreria")
|
|
REGGAETON_DIR = LIBRERIA_DIR / "reggaeton"
|
|
REFERENCE_FILE = LIBRERIA_DIR / "reggaeton_ejemplo.mp3"
|
|
PROFILE_FILE = REGGAETON_DIR / ".user_sound_profile.json"
|
|
|
|
# Roles de samples soportados
|
|
SAMPLE_ROLES = ["kick", "snare", "clap", "hat_closed", "hat_open",
|
|
"bass", "synth", "fx", "perc", "drum_loop"]
|
|
|
|
|
|
@dataclass
|
|
class SpectralFingerprint:
|
|
"""Fingerprint espectral completo de un audio."""
|
|
bpm: float = 0.0
|
|
key: str = ""
|
|
energy_curve: List[float] = field(default_factory=list)
|
|
mfccs_mean: List[float] = field(default_factory=list)
|
|
spectral_centroid_mean: float = 0.0
|
|
onset_strength_mean: float = 0.0
|
|
duration: float = 0.0
|
|
sample_rate: int = 0
|
|
|
|
def to_dict(self) -> Dict[str, Any]:
|
|
return {
|
|
"bpm": self.bpm,
|
|
"key": self.key,
|
|
"energy_curve": self.energy_curve,
|
|
"mfccs_mean": self.mfccs_mean,
|
|
"spectral_centroid_mean": self.spectral_centroid_mean,
|
|
"onset_strength_mean": self.onset_strength_mean,
|
|
"duration": self.duration,
|
|
"sample_rate": self.sample_rate
|
|
}
|
|
|
|
@classmethod
|
|
def from_dict(cls, data: Dict[str, Any]) -> "SpectralFingerprint":
|
|
return cls(
|
|
bpm=data.get("bpm", 0.0),
|
|
key=data.get("key", ""),
|
|
energy_curve=data.get("energy_curve", []),
|
|
mfccs_mean=data.get("mfccs_mean", []),
|
|
spectral_centroid_mean=data.get("spectral_centroid_mean", 0.0),
|
|
onset_strength_mean=data.get("onset_strength_mean", 0.0),
|
|
duration=data.get("duration", 0.0),
|
|
sample_rate=data.get("sample_rate", 0)
|
|
)
|
|
|
|
|
|
@dataclass
|
|
class SampleMatch:
|
|
"""Resultado de comparación de un sample contra referencia."""
|
|
path: str
|
|
name: str
|
|
role: str
|
|
similarity_score: float
|
|
fingerprint: SpectralFingerprint
|
|
|
|
|
|
@dataclass
|
|
class UserSoundProfile:
|
|
"""Perfil de sonido personalizado del usuario."""
|
|
# Características promedio ponderadas
|
|
preferred_bpm: float = 0.0
|
|
preferred_key: str = ""
|
|
preferred_timbre: List[float] = field(default_factory=list)
|
|
characteristic_energy_curve: List[float] = field(default_factory=list)
|
|
|
|
# Roles más usados (ordenados por frecuencia)
|
|
preferred_roles: List[str] = field(default_factory=list)
|
|
|
|
# Metadata
|
|
created_from_reference: str = ""
|
|
total_matches_analyzed: int = 0
|
|
genre: str = "reggaeton"
|
|
|
|
# Matches más similares por rol
|
|
top_matches_by_role: Dict[str, List[Dict]] = field(default_factory=dict)
|
|
|
|
def to_dict(self) -> Dict[str, Any]:
|
|
return {
|
|
"preferred_bpm": self.preferred_bpm,
|
|
"preferred_key": self.preferred_key,
|
|
"preferred_timbre": self.preferred_timbre,
|
|
"characteristic_energy_curve": self.characteristic_energy_curve,
|
|
"preferred_roles": self.preferred_roles,
|
|
"created_from_reference": self.created_from_reference,
|
|
"total_matches_analyzed": self.total_matches_analyzed,
|
|
"genre": self.genre,
|
|
"top_matches_by_role": self.top_matches_by_role
|
|
}
|
|
|
|
@classmethod
|
|
def from_dict(cls, data: Dict[str, Any]) -> "UserSoundProfile":
|
|
return cls(
|
|
preferred_bpm=data.get("preferred_bpm", 0.0),
|
|
preferred_key=data.get("preferred_key", ""),
|
|
preferred_timbre=data.get("preferred_timbre", []),
|
|
characteristic_energy_curve=data.get("characteristic_energy_curve", []),
|
|
preferred_roles=data.get("preferred_roles", []),
|
|
created_from_reference=data.get("created_from_reference", ""),
|
|
total_matches_analyzed=data.get("total_matches_analyzed", 0),
|
|
genre=data.get("genre", "reggaeton"),
|
|
top_matches_by_role=data.get("top_matches_by_role", {})
|
|
)
|
|
|
|
|
|
class AudioAnalyzer:
|
|
"""Analiza archivos de audio y extrae fingerprints espectrales."""
|
|
|
|
def __init__(self):
|
|
self._librosa_available = self._check_librosa()
|
|
|
|
def _check_librosa(self) -> bool:
|
|
"""Verifica si librosa está disponible."""
|
|
try:
|
|
import librosa
|
|
import librosa.display
|
|
return True
|
|
except ImportError:
|
|
logger.warning("librosa no disponible. Usando modo simulado.")
|
|
return False
|
|
|
|
def analyze_file(self, file_path: str) -> Optional[SpectralFingerprint]:
|
|
"""
|
|
Analiza un archivo de audio y extrae su fingerprint espectral.
|
|
|
|
Args:
|
|
file_path: Ruta al archivo de audio
|
|
|
|
Returns:
|
|
SpectralFingerprint con todas las características extraídas
|
|
"""
|
|
if not os.path.exists(file_path):
|
|
logger.error("Archivo no encontrado: %s", file_path)
|
|
return None
|
|
|
|
if self._librosa_available:
|
|
return self._analyze_with_librosa(file_path)
|
|
else:
|
|
return self._generate_mock_fingerprint(file_path)
|
|
|
|
def _analyze_with_librosa(self, file_path: str) -> Optional[SpectralFingerprint]:
|
|
"""Análisis real usando librosa."""
|
|
try:
|
|
import librosa
|
|
import librosa.display
|
|
|
|
# Cargar audio
|
|
y, sr = librosa.load(file_path, sr=None)
|
|
duration = librosa.get_duration(y=y, sr=sr)
|
|
|
|
# 1. Detectar BPM
|
|
tempo, _ = librosa.beat.beat_track(y=y, sr=sr)
|
|
bpm = float(tempo) if isinstance(tempo, (int, float, np.number)) else 95.0
|
|
|
|
# 2. Detectar Key (simplificado - usa chroma)
|
|
chroma = librosa.feature.chroma_stft(y=y, sr=sr)
|
|
chroma_mean = np.mean(chroma, axis=1)
|
|
key_idx = np.argmax(chroma_mean)
|
|
keys = ['C', 'C#', 'D', 'D#', 'E', 'F', 'F#', 'G', 'G#', 'A', 'A#', 'B']
|
|
key = keys[key_idx] + "m" # Asumimos menor para reggaeton
|
|
|
|
# 3. Energy curve (RMS por segmentos de 1 segundo)
|
|
hop_length = 512
|
|
frame_length = sr # 1 segundo
|
|
rms = librosa.feature.rms(y=y, frame_length=frame_length, hop_length=hop_length)[0]
|
|
energy_curve = rms.tolist() if len(rms) > 0 else [0.5]
|
|
|
|
# Normalizar a 16 segmentos máximo
|
|
if len(energy_curve) > 16:
|
|
# Agrupar en 16 segmentos
|
|
segment_size = len(energy_curve) // 16
|
|
energy_curve = [
|
|
np.mean(energy_curve[i:i+segment_size])
|
|
for i in range(0, len(energy_curve), segment_size)
|
|
][:16]
|
|
|
|
# 4. MFCCs (timbre) - promedio
|
|
mfccs = librosa.feature.mfcc(y=y, sr=sr, n_mfcc=13)
|
|
mfccs_mean = np.mean(mfccs, axis=1).tolist()
|
|
|
|
# 5. Spectral centroid (brillo)
|
|
spectral_centroids = librosa.feature.spectral_centroid(y=y, sr=sr)[0]
|
|
spectral_centroid_mean = float(np.mean(spectral_centroids))
|
|
|
|
# 6. Onset strength (ritmo/percussividad)
|
|
onset_env = librosa.onset.onset_strength(y=y, sr=sr)
|
|
onset_strength_mean = float(np.mean(onset_env))
|
|
|
|
logger.info("Análisis completado: %s (BPM: %.1f, Key: %s)",
|
|
file_path, bpm, key)
|
|
|
|
return SpectralFingerprint(
|
|
bpm=bpm,
|
|
key=key,
|
|
energy_curve=energy_curve,
|
|
mfccs_mean=mfccs_mean,
|
|
spectral_centroid_mean=spectral_centroid_mean,
|
|
onset_strength_mean=onset_strength_mean,
|
|
duration=duration,
|
|
sample_rate=sr
|
|
)
|
|
|
|
except Exception as e:
|
|
logger.error("Error analizando %s: %s", file_path, e)
|
|
return self._generate_mock_fingerprint(file_path)
|
|
|
|
def _generate_mock_fingerprint(self, file_path: str) -> SpectralFingerprint:
|
|
"""Genera fingerprint simulado para pruebas sin librosa."""
|
|
import hashlib
|
|
|
|
# Generar valores deterministas basados en el nombre del archivo
|
|
name_hash = hashlib.md5(file_path.encode()).hexdigest()
|
|
|
|
# BPM entre 85-105 (típico reggaeton)
|
|
bpm = 85 + (int(name_hash[:4], 16) % 20)
|
|
|
|
# Key basada en hash
|
|
keys = ['Am', 'Dm', 'Gm', 'Cm', 'Em', 'Bm', 'Fm']
|
|
key = keys[int(name_hash[4:6], 16) % len(keys)]
|
|
|
|
# Energy curve simulado (16 segmentos)
|
|
np.random.seed(int(name_hash[:8], 16))
|
|
energy_curve = np.random.uniform(0.3, 0.9, 16).tolist()
|
|
|
|
# MFCCs simulados
|
|
mfccs_mean = np.random.uniform(-50, 50, 13).tolist()
|
|
|
|
return SpectralFingerprint(
|
|
bpm=float(bpm),
|
|
key=key,
|
|
energy_curve=energy_curve,
|
|
mfccs_mean=mfccs_mean,
|
|
spectral_centroid_mean=float(2000 + int(name_hash[6:10], 16) % 2000),
|
|
onset_strength_mean=float(0.3 + (int(name_hash[10:12], 16) % 70) / 100),
|
|
duration=30.0,
|
|
sample_rate=44100
|
|
)
|
|
|
|
|
|
class SimilarityEngine:
|
|
"""Calcula similitud entre fingerprints espectrales."""
|
|
|
|
def find_similar(self,
|
|
reference: SpectralFingerprint,
|
|
candidates: List[Tuple[str, SpectralFingerprint]],
|
|
top_k: int = 20) -> List[SampleMatch]:
|
|
"""
|
|
Encuentra los samples más similares a la referencia.
|
|
|
|
Args:
|
|
reference: Fingerprint de referencia
|
|
candidates: Lista de (path, fingerprint) a comparar
|
|
top_k: Número de resultados a retornar
|
|
|
|
Returns:
|
|
Lista de SampleMatch ordenados por similitud
|
|
"""
|
|
matches = []
|
|
|
|
for path, candidate_fp in candidates:
|
|
score = self._calculate_similarity(reference, candidate_fp)
|
|
|
|
# Determinar rol basado en path
|
|
role = self._guess_role_from_path(path)
|
|
name = os.path.basename(path)
|
|
|
|
matches.append(SampleMatch(
|
|
path=path,
|
|
name=name,
|
|
role=role,
|
|
similarity_score=score,
|
|
fingerprint=candidate_fp
|
|
))
|
|
|
|
# Ordenar por score descendente
|
|
matches.sort(key=lambda x: x.similarity_score, reverse=True)
|
|
|
|
return matches[:top_k]
|
|
|
|
def _calculate_similarity(self,
|
|
ref: SpectralFingerprint,
|
|
cand: SpectralFingerprint) -> float:
|
|
"""
|
|
Calcula score de similitud entre dos fingerprints.
|
|
Retorna valor entre 0.0 y 1.0.
|
|
"""
|
|
scores = []
|
|
weights = []
|
|
|
|
# 1. Similitud de BPM (weight: 0.25)
|
|
if ref.bpm > 0 and cand.bpm > 0:
|
|
bpm_diff = abs(ref.bpm - cand.bpm)
|
|
bpm_sim = max(0, 1 - (bpm_diff / 30)) # 30 BPM de tolerancia
|
|
scores.append(bpm_sim)
|
|
weights.append(0.25)
|
|
|
|
# 2. Similitud de Key (weight: 0.15)
|
|
if ref.key and cand.key:
|
|
key_sim = 1.0 if ref.key == cand.key else 0.5 if ref.key[0] == cand.key[0] else 0.0
|
|
scores.append(key_sim)
|
|
weights.append(0.15)
|
|
|
|
# 3. Similitud de Energy Curve (weight: 0.25)
|
|
if ref.energy_curve and cand.energy_curve:
|
|
# Interpolar a mismo tamaño
|
|
min_len = min(len(ref.energy_curve), len(cand.energy_curve))
|
|
ref_curve = np.array(ref.energy_curve[:min_len])
|
|
cand_curve = np.array(cand.energy_curve[:min_len])
|
|
|
|
# Correlación de Pearson
|
|
if len(ref_curve) > 1:
|
|
corr = np.corrcoef(ref_curve, cand_curve)[0, 1]
|
|
if not np.isnan(corr):
|
|
energy_sim = (corr + 1) / 2 # Normalizar a 0-1
|
|
scores.append(energy_sim)
|
|
weights.append(0.25)
|
|
|
|
# 4. Similitud de Timbre (MFCCs) (weight: 0.20)
|
|
if ref.mfccs_mean and cand.mfccs_mean:
|
|
ref_mfccs = np.array(ref.mfccs_mean)
|
|
cand_mfccs = np.array(cand.mfccs_mean)
|
|
|
|
# Distancia euclidiana normalizada
|
|
distance = np.linalg.norm(ref_mfccs - cand_mfccs)
|
|
max_dist = np.linalg.norm(np.abs(ref_mfccs) + 100) # Estimación de max
|
|
timbre_sim = max(0, 1 - (distance / max_dist))
|
|
scores.append(timbre_sim)
|
|
weights.append(0.20)
|
|
|
|
# 5. Similitud de Spectral Centroid (weight: 0.10)
|
|
if ref.spectral_centroid_mean > 0 and cand.spectral_centroid_mean > 0:
|
|
sc_diff = abs(ref.spectral_centroid_mean - cand.spectral_centroid_mean)
|
|
sc_max = max(ref.spectral_centroid_mean, cand.spectral_centroid_mean)
|
|
sc_sim = max(0, 1 - (sc_diff / sc_max)) if sc_max > 0 else 0.5
|
|
scores.append(sc_sim)
|
|
weights.append(0.10)
|
|
|
|
# 6. Similitud de Onset Strength (weight: 0.05)
|
|
if ref.onset_strength_mean > 0 and cand.onset_strength_mean > 0:
|
|
os_diff = abs(ref.onset_strength_mean - cand.onset_strength_mean)
|
|
os_max = max(ref.onset_strength_mean, cand.onset_strength_mean)
|
|
os_sim = max(0, 1 - (os_diff / os_max)) if os_max > 0 else 0.5
|
|
scores.append(os_sim)
|
|
weights.append(0.05)
|
|
|
|
# Calcular promedio ponderado
|
|
if not scores:
|
|
return 0.5
|
|
|
|
total_weight = sum(weights)
|
|
weighted_score = sum(s * w for s, w in zip(scores, weights)) / total_weight
|
|
|
|
return float(weighted_score)
|
|
|
|
def _guess_role_from_path(self, path: str) -> str:
|
|
"""Infiere el rol del sample basado en su path."""
|
|
lower = path.lower()
|
|
|
|
if "kick" in lower:
|
|
return "kick"
|
|
if "snare" in lower:
|
|
return "snare"
|
|
if "clap" in lower:
|
|
return "clap"
|
|
if "hi-hat" in lower or "hihat" in lower:
|
|
return "hat_closed"
|
|
if "bass" in lower:
|
|
return "bass"
|
|
if "fx" in lower:
|
|
return "fx"
|
|
if "perc" in lower:
|
|
return "perc"
|
|
if "drumloop" in lower or "drum_loop" in lower:
|
|
return "drum_loop"
|
|
if "oneshot" in lower or "synth" in lower:
|
|
return "synth"
|
|
|
|
return "synth" # Default
|
|
|
|
|
|
class ReferenceMatcher:
|
|
"""
|
|
Matcher principal que analiza referencias y genera perfiles de usuario.
|
|
"""
|
|
|
|
def __init__(self,
|
|
reference_path: Optional[str] = None,
|
|
library_path: Optional[str] = None,
|
|
profile_path: Optional[str] = None):
|
|
self.reference_path = reference_path or str(REFERENCE_FILE)
|
|
self.library_path = library_path or str(REGGAETON_DIR)
|
|
self.profile_path = profile_path or str(PROFILE_FILE)
|
|
|
|
self.analyzer = AudioAnalyzer()
|
|
self.similarity = SimilarityEngine()
|
|
|
|
self._reference_fingerprint: Optional[SpectralFingerprint] = None
|
|
self._library_index: List[Tuple[str, SpectralFingerprint]] = []
|
|
self._profile: Optional[UserSoundProfile] = None
|
|
|
|
def analyze_reference(self) -> Optional[SpectralFingerprint]:
|
|
"""
|
|
Analiza el archivo de referencia y retorna su fingerprint.
|
|
|
|
Returns:
|
|
SpectralFingerprint del archivo de referencia
|
|
"""
|
|
logger.info("Analizando referencia: %s", self.reference_path)
|
|
|
|
self._reference_fingerprint = self.analyzer.analyze_file(self.reference_path)
|
|
|
|
if self._reference_fingerprint:
|
|
logger.info("Referencia analizada - BPM: %.1f, Key: %s",
|
|
self._reference_fingerprint.bpm,
|
|
self._reference_fingerprint.key)
|
|
|
|
return self._reference_fingerprint
|
|
|
|
def index_library(self, force_reindex: bool = False) -> List[Tuple[str, SpectralFingerprint]]:
|
|
"""
|
|
Indexa toda la librería y extrae fingerprints.
|
|
|
|
Args:
|
|
force_reindex: Si True, reindexa aunque ya exista índice
|
|
|
|
Returns:
|
|
Lista de (path, fingerprint) de todos los samples
|
|
"""
|
|
if self._library_index and not force_reindex:
|
|
return self._library_index
|
|
|
|
logger.info("Indexando librería: %s", self.library_path)
|
|
|
|
self._library_index = []
|
|
library = Path(self.library_path)
|
|
|
|
if not library.is_dir():
|
|
logger.error("Librería no encontrada: %s", self.library_path)
|
|
return []
|
|
|
|
audio_extensions = ('.wav', '.aif', '.aiff', '.mp3', '.flac', '.ogg')
|
|
|
|
for root, _dirs, files in os.walk(library):
|
|
for filename in files:
|
|
if filename.lower().endswith(audio_extensions):
|
|
filepath = os.path.join(root, filename)
|
|
|
|
# Analizar sample
|
|
fingerprint = self.analyzer.analyze_file(filepath)
|
|
|
|
if fingerprint:
|
|
self._library_index.append((filepath, fingerprint))
|
|
logger.debug("Indexado: %s", filename)
|
|
|
|
logger.info("Librería indexada: %d samples", len(self._library_index))
|
|
return self._library_index
|
|
|
|
def find_similar_samples(self,
|
|
top_k: int = 50,
|
|
role_filter: Optional[str] = None) -> List[SampleMatch]:
|
|
"""
|
|
Encuentra los samples más similares a la referencia.
|
|
|
|
Args:
|
|
top_k: Número de samples a retornar
|
|
role_filter: Si se especifica, filtra por rol específico
|
|
|
|
Returns:
|
|
Lista de SampleMatch ordenados por similitud
|
|
"""
|
|
if not self._reference_fingerprint:
|
|
self.analyze_reference()
|
|
|
|
if not self._library_index:
|
|
self.index_library()
|
|
|
|
if not self._reference_fingerprint or not self._library_index:
|
|
logger.error("No se puede buscar similares: falta referencia o librería")
|
|
return []
|
|
|
|
# Filtrar por rol si es necesario
|
|
candidates = self._library_index
|
|
if role_filter:
|
|
candidates = [
|
|
(path, fp) for path, fp in candidates
|
|
if self.similarity._guess_role_from_path(path) == role_filter
|
|
]
|
|
|
|
logger.info("Buscando %d samples similares (filtro: %s)...",
|
|
top_k, role_filter or "ninguno")
|
|
|
|
matches = self.similarity.find_similar(
|
|
self._reference_fingerprint,
|
|
candidates,
|
|
top_k=top_k
|
|
)
|
|
|
|
return matches
|
|
|
|
def generate_user_profile(self,
|
|
top_matches_count: int = 100,
|
|
save: bool = True) -> UserSoundProfile:
|
|
"""
|
|
Genera el perfil de sonido del usuario basado en matches similares.
|
|
|
|
Args:
|
|
top_matches_count: Cuántos matches usar para el perfil
|
|
save: Si True, guarda el perfil en disco
|
|
|
|
Returns:
|
|
UserSoundProfile generado
|
|
"""
|
|
logger.info("Generando perfil de usuario...")
|
|
|
|
# Obtener matches
|
|
matches = self.find_similar_samples(top_k=top_matches_count)
|
|
|
|
if not matches:
|
|
logger.warning("No hay matches para generar perfil")
|
|
return UserSoundProfile()
|
|
|
|
# Calcular BPM preferido (promedio ponderado por similitud)
|
|
total_weight = sum(m.similarity_score for m in matches)
|
|
weighted_bpm = sum(m.fingerprint.bpm * m.similarity_score
|
|
for m in matches if m.fingerprint.bpm > 0)
|
|
preferred_bpm = weighted_bpm / total_weight if total_weight > 0 else 95.0
|
|
|
|
# Calcular Key preferida (moda)
|
|
keys = [m.fingerprint.key for m in matches if m.fingerprint.key]
|
|
preferred_key = Counter(keys).most_common(1)[0][0] if keys else "Am"
|
|
|
|
# Calcular Timbre promedio (MFCCs ponderados)
|
|
mfccs_list = []
|
|
weights = []
|
|
for m in matches:
|
|
if m.fingerprint.mfccs_mean:
|
|
mfccs_list.append(np.array(m.fingerprint.mfccs_mean))
|
|
weights.append(m.similarity_score)
|
|
|
|
if mfccs_list and weights:
|
|
weighted_mfccs = np.average(mfccs_list, axis=0, weights=weights)
|
|
preferred_timbre = weighted_mfccs.tolist()
|
|
else:
|
|
preferred_timbre = []
|
|
|
|
# Energy curve característico (promedio de los matches)
|
|
energy_curves = []
|
|
for m in matches:
|
|
if m.fingerprint.energy_curve:
|
|
energy_curves.append(np.array(m.fingerprint.energy_curve))
|
|
|
|
if energy_curves:
|
|
# Interpolar todos a 16 segmentos
|
|
interpolated = []
|
|
for ec in energy_curves:
|
|
if len(ec) < 16:
|
|
# Replicar para llegar a 16
|
|
repeated = np.repeat(ec, 16 // len(ec) + 1)[:16]
|
|
interpolated.append(repeated)
|
|
else:
|
|
interpolated.append(ec[:16])
|
|
|
|
char_energy_curve = np.mean(interpolated, axis=0).tolist()
|
|
else:
|
|
char_energy_curve = [0.5] * 16
|
|
|
|
# Roles más usados
|
|
role_counts = Counter(m.role for m in matches)
|
|
preferred_roles = [role for role, _ in role_counts.most_common()]
|
|
|
|
# Top matches por rol
|
|
top_by_role: Dict[str, List[Dict]] = {}
|
|
for role in SAMPLE_ROLES:
|
|
role_matches = [m for m in matches if m.role == role][:10]
|
|
if role_matches:
|
|
top_by_role[role] = [
|
|
{
|
|
"path": m.path,
|
|
"name": m.name,
|
|
"similarity_score": m.similarity_score,
|
|
"bpm": m.fingerprint.bpm,
|
|
"key": m.fingerprint.key
|
|
}
|
|
for m in role_matches
|
|
]
|
|
|
|
# Crear perfil
|
|
profile = UserSoundProfile(
|
|
preferred_bpm=preferred_bpm,
|
|
preferred_key=preferred_key,
|
|
preferred_timbre=preferred_timbre,
|
|
characteristic_energy_curve=char_energy_curve,
|
|
preferred_roles=preferred_roles,
|
|
created_from_reference=self.reference_path,
|
|
total_matches_analyzed=len(matches),
|
|
genre="reggaeton",
|
|
top_matches_by_role=top_by_role
|
|
)
|
|
|
|
self._profile = profile
|
|
|
|
if save:
|
|
self._save_profile(profile)
|
|
|
|
logger.info("Perfil generado - BPM: %.1f, Key: %s, Roles: %s",
|
|
preferred_bpm, preferred_key, preferred_roles[:5])
|
|
|
|
return profile
|
|
|
|
def _save_profile(self, profile: UserSoundProfile) -> bool:
|
|
"""Guarda el perfil en disco."""
|
|
try:
|
|
profile_data = profile.to_dict()
|
|
|
|
with open(self.profile_path, 'w', encoding='utf-8') as f:
|
|
json.dump(profile_data, f, indent=2, ensure_ascii=False)
|
|
|
|
logger.info("Perfil guardado en: %s", self.profile_path)
|
|
return True
|
|
|
|
except Exception as e:
|
|
logger.error("Error guardando perfil: %s", e)
|
|
return False
|
|
|
|
def load_profile(self) -> Optional[UserSoundProfile]:
|
|
"""
|
|
Carga el perfil desde disco.
|
|
|
|
Returns:
|
|
UserSoundProfile o None si no existe
|
|
"""
|
|
if not os.path.exists(self.profile_path):
|
|
logger.info("No existe perfil guardado en: %s", self.profile_path)
|
|
return None
|
|
|
|
try:
|
|
with open(self.profile_path, 'r', encoding='utf-8') as f:
|
|
data = json.load(f)
|
|
|
|
self._profile = UserSoundProfile.from_dict(data)
|
|
logger.info("Perfil cargado desde: %s", self.profile_path)
|
|
return self._profile
|
|
|
|
except Exception as e:
|
|
logger.error("Error cargando perfil: %s", e)
|
|
return None
|
|
|
|
def get_user_profile(self) -> UserSoundProfile:
|
|
"""
|
|
Obtiene el perfil del usuario, cargándolo o generándolo si no existe.
|
|
|
|
Returns:
|
|
UserSoundProfile del usuario
|
|
"""
|
|
# Intentar cargar
|
|
profile = self.load_profile()
|
|
|
|
if profile:
|
|
self._profile = profile
|
|
return profile
|
|
|
|
# Generar nuevo
|
|
logger.info("Generando nuevo perfil de usuario...")
|
|
return self.generate_user_profile()
|
|
|
|
def get_recommended_samples(self,
|
|
role: str,
|
|
count: int = 5,
|
|
bpm_tolerance: float = 5.0) -> List[Dict[str, Any]]:
|
|
"""
|
|
Retorna samples recomendados basados en el perfil del usuario.
|
|
|
|
Args:
|
|
role: Rol del sample deseado (kick, snare, bass, etc.)
|
|
count: Número de samples a retornar
|
|
bpm_tolerance: Tolerancia de BPM para filtrar
|
|
|
|
Returns:
|
|
Lista de diccionarios con información de samples recomendados
|
|
"""
|
|
# Asegurar que tenemos perfil
|
|
if not self._profile:
|
|
self.get_user_profile()
|
|
|
|
profile = self._profile
|
|
if not profile:
|
|
logger.warning("No se pudo obtener perfil, usando recomendaciones genéricas")
|
|
# Fallback: buscar similares sin perfil
|
|
matches = self.find_similar_samples(top_k=count * 3, role_filter=role)
|
|
return [
|
|
{
|
|
"path": m.path,
|
|
"name": m.name,
|
|
"role": m.role,
|
|
"similarity_score": m.similarity_score,
|
|
"bpm": m.fingerprint.bpm,
|
|
"key": m.fingerprint.key,
|
|
"reason": "Similitud directa con referencia"
|
|
}
|
|
for m in matches[:count]
|
|
]
|
|
|
|
# Buscar en top_matches_by_role del perfil
|
|
if role in profile.top_matches_by_role:
|
|
matches = profile.top_matches_by_role[role]
|
|
|
|
# Filtrar por BPM dentro de tolerancia
|
|
filtered = [
|
|
m for m in matches
|
|
if abs(m.get("bpm", 0) - profile.preferred_bpm) <= bpm_tolerance
|
|
]
|
|
|
|
# Si no hay suficientes con BPM cercano, usar todos
|
|
if len(filtered) < count:
|
|
filtered = matches
|
|
|
|
recommendations = filtered[:count]
|
|
|
|
return [
|
|
{
|
|
"path": r["path"],
|
|
"name": r["name"],
|
|
"role": role,
|
|
"similarity_score": r["similarity_score"],
|
|
"bpm": r.get("bpm", 0),
|
|
"key": r.get("key", ""),
|
|
"reason": f"Match con perfil (Key: {profile.preferred_key}, BPM: {profile.preferred_bpm:.1f})"
|
|
}
|
|
for r in recommendations
|
|
]
|
|
|
|
# Si no hay matches en el perfil para este rol, buscar en tiempo real
|
|
logger.info("No hay matches en perfil para '%s', buscando en librería...", role)
|
|
matches = self.find_similar_samples(top_k=count * 2, role_filter=role)
|
|
|
|
return [
|
|
{
|
|
"path": m.path,
|
|
"name": m.name,
|
|
"role": m.role,
|
|
"similarity_score": m.similarity_score,
|
|
"bpm": m.fingerprint.bpm,
|
|
"key": m.fingerprint.key,
|
|
"reason": "Búsqueda en tiempo real"
|
|
}
|
|
for m in matches[:count]
|
|
]
|
|
|
|
def get_profile_summary(self) -> Dict[str, Any]:
|
|
"""
|
|
Retorna resumen del perfil para debugging/visualización.
|
|
|
|
Returns:
|
|
Diccionario con resumen del perfil
|
|
"""
|
|
if not self._profile:
|
|
self.get_user_profile()
|
|
|
|
if not self._profile:
|
|
return {"error": "No se pudo generar perfil"}
|
|
|
|
p = self._profile
|
|
|
|
return {
|
|
"preferred_bpm": round(p.preferred_bpm, 1),
|
|
"preferred_key": p.preferred_key,
|
|
"characteristic_energy_curve": [round(x, 3) for x in p.characteristic_energy_curve[:8]],
|
|
"preferred_roles": p.preferred_roles[:5],
|
|
"top_matches_by_role_count": {
|
|
role: len(matches)
|
|
for role, matches in p.top_matches_by_role.items()
|
|
},
|
|
"total_matches_analyzed": p.total_matches_analyzed,
|
|
"created_from": p.created_from_reference,
|
|
"genre": p.genre
|
|
}
|
|
|
|
|
|
# Funciones de conveniencia globales
|
|
_matcher: Optional[ReferenceMatcher] = None
|
|
|
|
|
|
def get_matcher(reference_path: Optional[str] = None,
|
|
library_path: Optional[str] = None) -> ReferenceMatcher:
|
|
"""Obtiene instancia global del matcher."""
|
|
global _matcher
|
|
if _matcher is None:
|
|
_matcher = ReferenceMatcher(reference_path, library_path)
|
|
return _matcher
|
|
|
|
|
|
def get_user_profile(reference_path: Optional[str] = None,
|
|
library_path: Optional[str] = None) -> Dict[str, Any]:
|
|
"""
|
|
Función principal: obtiene o genera el perfil del usuario.
|
|
|
|
Args:
|
|
reference_path: Ruta al archivo de referencia (opcional)
|
|
library_path: Ruta a la librería de samples (opcional)
|
|
|
|
Returns:
|
|
Diccionario con el perfil del usuario
|
|
"""
|
|
matcher = get_matcher(reference_path, library_path)
|
|
profile = matcher.get_user_profile()
|
|
return profile.to_dict()
|
|
|
|
|
|
def get_recommended_samples(role: str,
|
|
count: int = 5,
|
|
reference_path: Optional[str] = None,
|
|
library_path: Optional[str] = None) -> List[Dict[str, Any]]:
|
|
"""
|
|
Obtiene samples recomendados para un rol específico.
|
|
|
|
Args:
|
|
role: Rol del sample (kick, snare, bass, synth, etc.)
|
|
count: Número de samples a retornar
|
|
reference_path: Ruta al archivo de referencia (opcional)
|
|
library_path: Ruta a la librería (opcional)
|
|
|
|
Returns:
|
|
Lista de samples recomendados
|
|
"""
|
|
matcher = get_matcher(reference_path, library_path)
|
|
return matcher.get_recommended_samples(role, count)
|
|
|
|
|
|
def analyze_reference(file_path: str) -> Optional[Dict[str, Any]]:
|
|
"""
|
|
Analiza un archivo de referencia y retorna su fingerprint.
|
|
|
|
Args:
|
|
file_path: Ruta al archivo de audio
|
|
|
|
Returns:
|
|
Diccionario con el fingerprint o None si falla
|
|
"""
|
|
analyzer = AudioAnalyzer()
|
|
fingerprint = analyzer.analyze_file(file_path)
|
|
|
|
if fingerprint:
|
|
return fingerprint.to_dict()
|
|
|
|
return None
|
|
|
|
|
|
def refresh_profile() -> Dict[str, Any]:
|
|
"""
|
|
Fuerza la regeneración del perfil del usuario.
|
|
|
|
Returns:
|
|
Nuevo perfil generado
|
|
"""
|
|
global _matcher
|
|
_matcher = None # Reset para forzar regeneración
|
|
|
|
matcher = get_matcher()
|
|
profile = matcher.generate_user_profile(save=True)
|
|
|
|
return profile.to_dict()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
# Test del módulo
|
|
logging.basicConfig(level=logging.INFO)
|
|
|
|
print("=" * 60)
|
|
print("Reference Matcher - Test")
|
|
print("=" * 60)
|
|
|
|
# Test 1: Analizar referencia
|
|
print("\n1. Analizando referencia...")
|
|
matcher = ReferenceMatcher()
|
|
ref_fp = matcher.analyze_reference()
|
|
|
|
if ref_fp:
|
|
print(f" BPM: {ref_fp.bpm}")
|
|
print(f" Key: {ref_fp.key}")
|
|
print(f" Duration: {ref_fp.duration:.2f}s")
|
|
|
|
# Test 2: Indexar librería
|
|
print("\n2. Indexando librería...")
|
|
library = matcher.index_library()
|
|
print(f" Samples indexados: {len(library)}")
|
|
|
|
# Test 3: Generar perfil
|
|
print("\n3. Generando perfil de usuario...")
|
|
profile = matcher.generate_user_profile(top_matches_count=30)
|
|
print(f" Preferred BPM: {profile.preferred_bpm:.1f}")
|
|
print(f" Preferred Key: {profile.preferred_key}")
|
|
print(f" Preferred Roles: {profile.preferred_roles[:3]}")
|
|
|
|
# Test 4: Recomendaciones
|
|
print("\n4. Obteniendo recomendaciones...")
|
|
for role in ["kick", "snare", "bass"]:
|
|
recs = matcher.get_recommended_samples(role, count=2)
|
|
print(f" {role}: {[r['name'] for r in recs]}")
|
|
|
|
print("\n" + "=" * 60)
|
|
print("Test completado!")
|
|
print("=" * 60)
|