feat: BPM and musical key detection for all clips
Backend:
- New analysis.py module using librosa for BPM + key detection
- detect_bpm(): librosa.beat.beat_track() returns tempo
- detect_key(): Krumhansl-Schmuckler key profiles on chroma features
- analyze_clip(): combined call returning {bpm, key}
- New dependency: librosa==0.10.2 + soundfile==0.12.1
- Database migrations: bpm (REAL) and key (TEXT) columns added
- New set_analysis() function in database.py
- New endpoint: GET /api/clips/{id}/analysis
- Returns cached analysis if available
- Computes on-demand if missing, caches result
- 400 if clip not ready, 404 if no audio file
Frontend:
- Clip interface extended with bpm?: number | null, key?: string | null
- Gallery: BPM and key badges shown in ribbon and tile views
- BPM badge: lyria-accent background
- Key badge: white/5 background
- DetailPanel: BPM and key shown in metadata grid when available
- App.tsx: handleSelectClip() fetches analysis on-demand when opening
detail panel if not already cached
Docs:
- Architecture KB updated with Audio Analysis section
- Tasks KB: T-028 marked In Progress → will mark Done after commit
This commit is contained in:
1 parent
2d47f77820
commit
1e21b7dcc8
7 files changed
+187
-3
No files matched your search
@@ -6,3 +6,5 @@ pydantic>=2.10.0
|
||||
python-multipart==0.0.12
|
||||
python-dotenv==1.0.0
|
||||
numpy==2.2.0
|
||||
librosa==0.10.2
|
||||
soundfile==0.12.1
|
||||
@@ -0,0 +1,82 @@
|
||||
"""Audio analysis: BPM and musical key detection using librosa."""
|
||||
|
||||
import librosa
|
||||
import numpy as np
|
||||
from pathlib import Path
|
||||
|
||||
# Krumhansl-Schmuckler key profiles (normalized)
|
||||
# Each profile is a 12-element vector representing the strength of each pitch class
|
||||
# in the key. Index 0 = C, 1 = C#, ..., 11 = B
|
||||
|
||||
MAJOR_PROFILE = np.array([
|
||||
6.35, 2.23, 3.48, 2.33, 4.38, 4.09, 2.52, 5.19, 2.39, 3.66, 2.29, 2.88
|
||||
])
|
||||
|
||||
MINOR_PROFILE = np.array([
|
||||
6.33, 2.68, 3.52, 5.38, 2.60, 3.53, 2.54, 4.75, 3.98, 2.69, 3.34, 3.17
|
||||
])
|
||||
|
||||
KEY_NAMES = [
|
||||
'C major', 'C# major', 'D major', 'D# major', 'E major', 'F major',
|
||||
'F# major', 'G major', 'G# major', 'A major', 'A# major', 'B major',
|
||||
'C minor', 'C# minor', 'D minor', 'D# minor', 'E minor', 'F minor',
|
||||
'F# minor', 'G minor', 'G# minor', 'A minor', 'A# minor', 'B minor',
|
||||
]
|
||||
|
||||
|
||||
def detect_bpm(audio_path: str | Path) -> float:
|
||||
"""Detect BPM (tempo) of an audio file using librosa."""
|
||||
y, sr = librosa.load(str(audio_path), sr=None)
|
||||
tempo, _ = librosa.beat.beat_track(y=y, sr=sr)
|
||||
# librosa returns tempo as a numpy array or scalar depending on version
|
||||
if isinstance(tempo, np.ndarray):
|
||||
return float(tempo[0])
|
||||
return float(tempo)
|
||||
|
||||
|
||||
def detect_key(audio_path: str | Path) -> str:
|
||||
"""Detect musical key using Krumhansl-Schmuckler key profiles."""
|
||||
y, sr = librosa.load(str(audio_path), sr=None)
|
||||
|
||||
# Extract chroma features
|
||||
chroma = librosa.feature.chroma_stft(y=y, sr=sr)
|
||||
|
||||
# Average chroma across time
|
||||
chroma_avg = np.mean(chroma, axis=1)
|
||||
|
||||
# Normalize
|
||||
chroma_avg = chroma_avg / (np.sum(chroma_avg) + 1e-10)
|
||||
|
||||
# Correlate with all 24 key profiles (12 major + 12 minor, each rotated)
|
||||
correlations = []
|
||||
|
||||
for i in range(12):
|
||||
# Rotate profiles to test each tonic
|
||||
major_rotated = np.roll(MAJOR_PROFILE, i)
|
||||
minor_rotated = np.roll(MINOR_PROFILE, i)
|
||||
|
||||
# Pearson correlation
|
||||
major_corr = np.corrcoef(chroma_avg, major_rotated)[0, 1]
|
||||
minor_corr = np.corrcoef(chroma_avg, minor_rotated)[0, 1]
|
||||
|
||||
correlations.append(major_corr)
|
||||
correlations.append(minor_corr)
|
||||
|
||||
# Find best match
|
||||
best_idx = int(np.argmax(correlations))
|
||||
return KEY_NAMES[best_idx]
|
||||
|
||||
|
||||
def analyze_clip(audio_path: str | Path) -> dict:
|
||||
"""Full analysis: BPM and key in one call."""
|
||||
audio_path = Path(audio_path)
|
||||
if not audio_path.exists():
|
||||
raise FileNotFoundError(f"Audio file not found: {audio_path}")
|
||||
|
||||
bpm = detect_bpm(audio_path)
|
||||
key = detect_key(audio_path)
|
||||
|
||||
return {
|
||||
'bpm': round(bpm, 1),
|
||||
'key': key,
|
||||
}
|
||||
@@ -68,6 +68,16 @@ async def init_db():
|
||||
await db.execute("ALTER TABLE clips ADD COLUMN artwork_filename TEXT")
|
||||
except Exception:
|
||||
pass # Column already exists
|
||||
# Migration: add bpm column if it doesn't exist
|
||||
try:
|
||||
await db.execute("ALTER TABLE clips ADD COLUMN bpm REAL")
|
||||
except Exception:
|
||||
pass # Column already exists
|
||||
# Migration: add key column if it doesn't exist
|
||||
try:
|
||||
await db.execute("ALTER TABLE clips ADD COLUMN key TEXT")
|
||||
except Exception:
|
||||
pass # Column already exists
|
||||
await db.commit()
|
||||
|
||||
async def create_clip(prompt: str, model: str, duration: int) -> str:
|
||||
@@ -148,6 +158,14 @@ async def remove_artwork(clip_id: str):
|
||||
)
|
||||
await db.commit()
|
||||
|
||||
async def set_analysis(clip_id: str, bpm: float, key: str):
|
||||
async with aiosqlite.connect(DB_PATH) as db:
|
||||
await db.execute(
|
||||
"UPDATE clips SET bpm = ?, key = ? WHERE id = ?",
|
||||
(bpm, key, clip_id)
|
||||
)
|
||||
await db.commit()
|
||||
|
||||
async def delete_clip(clip_id: str):
|
||||
async with aiosqlite.connect(DB_PATH) as db:
|
||||
await db.execute("DELETE FROM clips WHERE id = ?", (clip_id,))
|
||||
|
||||
+28
-1
@@ -10,9 +10,10 @@ from dotenv import load_dotenv
|
||||
load_dotenv(Path(__file__).parent.parent / ".env")
|
||||
|
||||
from models import GenerateRequest, GenerateResponse, ClipList, RenameRequest, LYRIA_MODELS, DEFAULT_MODEL
|
||||
from database import init_db, create_clip, get_clips, get_clip, delete_clip, update_clip_status, rename_clip, set_artwork, remove_artwork, delete_all_clips
|
||||
from database import init_db, create_clip, get_clips, get_clip, delete_clip, update_clip_status, rename_clip, set_artwork, remove_artwork, delete_all_clips, set_analysis
|
||||
from lyria_client import generate_clip, save_clip_file
|
||||
from waveform import generate_waveform_svg, get_waveform_path
|
||||
from analysis import analyze_clip
|
||||
|
||||
app = FastAPI(title="Lyria Studio", version="0.1.0")
|
||||
|
||||
@@ -228,6 +229,32 @@ async def rename_clip_endpoint(clip_id: str, request: RenameRequest):
|
||||
|
||||
# ─── Admin Endpoints ──────────────────────────────────────────────
|
||||
|
||||
@app.get("/api/clips/{clip_id}/analysis")
|
||||
async def get_clip_analysis(clip_id: str):
|
||||
"""Get BPM and key analysis for a clip. Computes on-demand if not cached."""
|
||||
clip = await get_clip(clip_id)
|
||||
if not clip:
|
||||
raise HTTPException(status_code=404, detail="Clip not found")
|
||||
if clip.get("status") != "ready":
|
||||
raise HTTPException(status_code=400, detail="Clip is not ready for analysis")
|
||||
|
||||
# Return cached analysis if available
|
||||
if clip.get("bpm") and clip.get("key"):
|
||||
return {"bpm": clip["bpm"], "key": clip["key"]}
|
||||
|
||||
# Compute analysis
|
||||
mp3_path = CLIPS_DIR / clip["filename"]
|
||||
if not mp3_path.exists():
|
||||
raise HTTPException(status_code=404, detail="Audio file not found")
|
||||
|
||||
try:
|
||||
result = analyze_clip(mp3_path)
|
||||
# Cache in DB
|
||||
await set_analysis(clip_id, result["bpm"], result["key"])
|
||||
return result
|
||||
except Exception as e:
|
||||
raise HTTPException(status_code=500, detail=f"Analysis failed: {str(e)}")
|
||||
|
||||
@app.post("/api/admin/clear-waveforms")
|
||||
async def clear_waveforms():
|
||||
"""Delete all cached waveform SVGs. They regenerate on next view."""
|
||||
|
||||
Reference in new issue
Block a user