d339ba588d
- Create lightsync/audio/features.py with extract_features and extract_features_async - Create lightsync/audio/segments.py with detect_segments via agglomerative clustering - Add /api/audio/segments and /api/audio/features endpoints to audio.py - Initialize app.state.segments cache in main.py lifespan
68 lines
2.3 KiB
Python
68 lines
2.3 KiB
Python
"""Structural segmentation via librosa agglomerative clustering (Phase 6).
|
|
|
|
detect_segments(path, n_segments=None) -> [{"start": float, "end": float, "label": str}, ...]
|
|
|
|
Uses chroma features + recurrence matrix + agglomerative clustering.
|
|
Target 4-12 sections for a typical 3-5 minute song.
|
|
"""
|
|
from __future__ import annotations
|
|
import asyncio
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
|
|
def detect_segments(path: str, n_segments: int | None = None) -> list[dict[str, Any]]:
|
|
"""Detect structural sections using agglomerative clustering on chroma features.
|
|
|
|
Args:
|
|
path: Absolute path to audio file.
|
|
n_segments: Number of segments (default: auto based on duration, ~1 per 30s).
|
|
|
|
Returns:
|
|
List of dicts: [{"start": float, "end": float, "label": "SEC 1"}, ...]
|
|
"""
|
|
import librosa
|
|
import numpy as np
|
|
|
|
p = Path(path)
|
|
if not p.exists():
|
|
raise FileNotFoundError(f"Audio file not found: {path}")
|
|
|
|
y, sr = librosa.load(path, sr=None, mono=True)
|
|
duration = len(y) / sr
|
|
|
|
# Auto-select segment count if not specified: ~1 per 30s, clamped to 4-12
|
|
if n_segments is None:
|
|
n_segments = max(4, min(12, int(duration / 30)))
|
|
|
|
# Build chroma feature matrix
|
|
chroma = librosa.feature.chroma_cqt(y=y, sr=sr, bins_per_octave=36)
|
|
|
|
# Recurrence matrix — similarity structure
|
|
R = librosa.segment.recurrence_matrix(chroma, width=3, mode='affinity', sym=True)
|
|
|
|
# Filter recurrence matrix to enhance diagonal consistency
|
|
R_filtered = librosa.segment.path_enhance(R, 15)
|
|
|
|
# Agglomerative clustering — returns boundary frame indices
|
|
bounds = librosa.segment.agglomerative(R_filtered, k=n_segments)
|
|
bound_times = librosa.frames_to_time(bounds, sr=sr).tolist()
|
|
|
|
# Build section list: [{start, end, label}, ...]
|
|
# Pitfall 2 from RESEARCH: agglomerative returns k boundaries, yielding k+1 sections
|
|
all_times = [0.0] + bound_times + [duration]
|
|
sections = []
|
|
for i in range(len(all_times) - 1):
|
|
sections.append({
|
|
"start": all_times[i],
|
|
"end": all_times[i + 1],
|
|
"label": f"SEC {i + 1}",
|
|
})
|
|
|
|
return sections
|
|
|
|
|
|
async def detect_segments_async(path: str, n_segments: int | None = None) -> list[dict[str, Any]]:
|
|
"""Async wrapper — runs in thread pool."""
|
|
return await asyncio.to_thread(detect_segments, path, n_segments)
|